"""/entries reads: list (sort + pagination), entries_filter_totals
(rpcs §2.1) and entries_filter_options (§2.2) equivalents. All
aggregation happens in SQL; every query is scope()-transformed.

Deliberate fixes over the legacy RPCs (binding, rpcs §8.3):
  - the vestigial `billout_seconds == source_seconds` duplicate column
    is NOT ported (one hours figure: source seconds);
  - the options arrays normalize SQL NULL -> [];
  - `?invoice=` applies to list AND totals alike (kills the B13 wart).
"""

from __future__ import annotations

from dataclasses import dataclass
from decimal import Decimal

from sqlalchemy import distinct, func, select
from sqlalchemy.orm import Session

from app.models import ImportBatch, Invoice, InvoiceLine, TimeEntry, User
from app.services.authz.scope import scope
from app.services.reporting.filters import (
    EntryFilters,
    EntryJoins,
    apply_filters,
    blocked_reason,
    status_case,
    status_sort_rank,
)
from app.services.reporting.money import money_from_sum, money_or_none

PAGE_SIZE = 100

#: sort key -> expression builder (entries.md B16, FK edition; the
#: legacy dead keys and the transferred_at status wart do not port).
_STATUS_RANK = {"blocked": 0, "pending": 1, "applied": 2}


def _sort_expr(joins: EntryJoins, key: str):
    mapping = {
        "date": TimeEntry.start_at,
        "user": joins.user_expr,
        "operator": joins.operator_expr,
        "client": joins.client_expr,
        "project": joins.project_expr,
        "description": TimeEntry.description,
        "source_hrs": TimeEntry.duration_seconds,
        "cost": TimeEntry.cost_amount,
        "billout_amount": TimeEntry.billout_amount,
        # ONE eligibility predicate (T-PHX-009): the status sort rank reuses
        # filters.status_sort_rank instead of re-inlining the conjunction.
        "status": status_sort_rank(),
    }
    return mapping.get(key, TimeEntry.start_at)


@dataclass(frozen=True)
class EntriesPage:
    rows: list[dict]
    total_count: int
    page: int
    page_size: int
    total_pages: int


def _row_select(joins: EntryJoins):
    """The one row-shaped select — shared by the page list and the
    single-row refetch mutations return (one code path, one shape)."""
    invoice_line = InvoiceLine.__table__.alias("il")
    invoice = Invoice.__table__.alias("inv")
    return (
        joins.select_from(
            TimeEntry.id,
            TimeEntry.start_at,
            TimeEntry.end_at,
            joins.user_expr.label("worker"),
            TimeEntry.source_user_email,
            TimeEntry.source_user_name,
            joins.operator_expr.label("operator"),
            joins.client_expr.label("client"),
            joins.project_expr.label("project"),
            TimeEntry.project_id,
            TimeEntry.worker_party_id,
            TimeEntry.description,
            TimeEntry.billable,
            TimeEntry.duration_seconds,
            TimeEntry.cost_amount,
            TimeEntry.cost_currency,
            TimeEntry.billout_amount,
            TimeEntry.billout_currency,
            status_case(),
            TimeEntry.invoice_line_id,
            TimeEntry.import_id,
            invoice.c.name.label("invoice_name"),
        )
        .outerjoin(invoice_line, invoice_line.c.id == TimeEntry.invoice_line_id)
        .outerjoin(invoice, invoice.c.id == invoice_line.c.invoice_id)
    )


def _shape_row(r, *, can_see_money: bool, zero_currency: str) -> dict:
    row = {
        "id": str(r["id"]),
        "start_at": r["start_at"].isoformat(sep=" ") if r["start_at"] else None,
        "end_at": r["end_at"].isoformat(sep=" ") if r["end_at"] else None,
        "worker": r["worker"],
        "worker_party_id": str(r["worker_party_id"]) if r["worker_party_id"] else None,
        "source_user_email": r["source_user_email"],
        "source_user_name": r["source_user_name"],
        "operator": r["operator"],
        "client": r["client"],
        "project": r["project"],
        "project_id": str(r["project_id"]) if r["project_id"] else None,
        "description": r["description"],
        "billable": r["billable"],
        "duration_seconds": r["duration_seconds"],
        "cost": _money_dict(r["cost_amount"], r["cost_currency"], zero_currency),
        "status": r["status"],
        "locked": r["invoice_line_id"] is not None,
        "invoice_name": r["invoice_name"],
        "import_id": str(r["import_id"]) if r["import_id"] else None,
        "blocked_reason": (
            blocked_reason(
                worker=r["worker"],
                operator=r["operator"],
                client=r["client"],
                project=r["project"],
                description=r["description"],
                end_at=r["end_at"],
                source_user_email=r["source_user_email"],
            )
            if r["status"] == "blocked"
            else None
        ),
    }
    if can_see_money:  # T-PRM-05: absent from the payload, not just the UI
        row["billout"] = _money_dict(
            r["billout_amount"], r["billout_currency"], zero_currency
        )
    return row


def fetch_entry_row(
    session: Session, actor, entry_id, *, zero_currency: str = "CAD"
) -> dict | None:
    """One shaped row through the SAME select/scope path as the list —
    what the write endpoints return so the SPA can update in place."""
    joins = EntryJoins()
    stmt = scope(actor, "entries:read", _row_select(joins)).where(TimeEntry.id == entry_id)
    r = session.execute(stmt).mappings().first()
    if r is None:
        return None
    return _shape_row(
        r, can_see_money=actor.allowed("can_see_income"), zero_currency=zero_currency
    )


def list_entries(
    session: Session,
    actor,
    f: EntryFilters,
    *,
    sort: str = "date",
    direction: str = "desc",
    page: int = 1,
    zero_currency: str = "CAD",
) -> EntriesPage:
    joins = EntryJoins()
    stmt = apply_filters(scope(actor, "entries:read", _row_select(joins)), joins, f)

    total_count = session.execute(
        select(func.count()).select_from(stmt.order_by(None).subquery())
    ).scalar_one()

    ascending = direction == "asc"
    order = _sort_expr(joins, sort)
    order = order.asc() if ascending else order.desc()
    stmt = stmt.order_by(
        order.nulls_last(), TimeEntry.start_at.desc(), TimeEntry.id
    )

    page = max(1, page)
    stmt = stmt.limit(PAGE_SIZE).offset((page - 1) * PAGE_SIZE)

    can_see_money = actor.allowed("can_see_income")
    rows = [
        _shape_row(r, can_see_money=can_see_money, zero_currency=zero_currency)
        for r in session.execute(stmt).mappings()
    ]

    return EntriesPage(
        rows=rows,
        total_count=total_count,
        page=page,
        page_size=PAGE_SIZE,
        total_pages=max(1, -(-total_count // PAGE_SIZE)),
    )


def _money_dict(amount, currency, zero_currency) -> dict | None:
    money = money_or_none(
        Decimal(amount) if amount is not None else None, currency, zero_currency
    )
    return money.as_dict() if money else None


def entry_totals(
    session: Session, actor, f: EntryFilters, *, zero_currency: str = "CAD"
) -> dict:
    """rpcs §2.1 equivalence: one aggregate row over the WHOLE filter
    match. Zero matching rows -> zeros. Single SQL pass; currency
    integrity asserted in the same pass."""
    joins = EntryJoins()
    stmt = apply_filters(
        scope(
            actor,
            "entries:read",
            joins.select_from(
                func.count().label("row_count"),
                func.coalesce(func.sum(TimeEntry.duration_seconds), 0).label("secs"),
                func.sum(TimeEntry.cost_amount).label("cost"),
                func.count(distinct(TimeEntry.cost_currency)).label("cost_ccys"),
                func.min(TimeEntry.cost_currency).label("cost_ccy"),
                func.sum(TimeEntry.billout_amount).label("billout"),
                func.count(distinct(TimeEntry.billout_currency)).label("billout_ccys"),
                func.min(TimeEntry.billout_currency).label("billout_ccy"),
            ),
        ),
        joins,
        f,
    )
    r = session.execute(stmt).mappings().one()
    cost = money_from_sum(
        r["cost"], r["cost_ccys"], r["cost_ccy"],
        context="entries totals: cost", zero_currency=zero_currency,
    )
    totals = {
        "row_count": r["row_count"],
        "source_seconds": int(r["secs"]),
        "cost": cost.as_dict(),
    }
    if actor.allowed("can_see_income"):
        billout = money_from_sum(
            r["billout"], r["billout_ccys"], r["billout_ccy"],
            context="entries totals: billout", zero_currency=zero_currency,
        )
        totals["billout"] = billout.as_dict()
    return totals


def entry_options(session: Session, actor, f: EntryFilters) -> dict:
    """rpcs §2.2 equivalence: cascading dropdown options — distinct,
    ascending, NULLs excluded, computed over the filtered + scoped set
    (so the dropdowns narrow as you filter). SQL NULL normalizes to [].

    Also returns the batches list (labels per entries.md B15) and, for
    can_transfer holders not viewing-as, the importer options."""

    def distinct_values(expr) -> list[str]:
        joins = EntryJoins()
        # rebuild the expression against THIS query's aliases
        column = expr(joins)
        stmt = apply_filters(
            scope(actor, "entries:read", joins.select_from(column.label("v")).distinct()),
            joins,
            f,
        ).where(column.is_not(None))
        return sorted(v for (v,) in session.execute(stmt).all())

    options = {
        "operators": distinct_values(lambda j: j.operator_expr),
        "clients": distinct_values(lambda j: j.client_expr),
        "projects": distinct_values(lambda j: j.project_expr),
        "users": distinct_values(lambda j: j.user_expr),
        "source_emails": distinct_values(lambda _j: TimeEntry.source_user_email),
    }

    batches_stmt = scope(
        actor,
        "imports:read",
        select(
            ImportBatch.id, ImportBatch.name, ImportBatch.filename, ImportBatch.imported_at
        ),
    ).order_by(ImportBatch.imported_at.desc())
    options["batches"] = [
        {
            "id": str(batch_id),
            "label": name
            or filename
            or (imported_at.date().isoformat() if imported_at else str(batch_id)[:8]),
        }
        for batch_id, name, filename, imported_at in session.execute(batches_stmt).all()
    ]

    # Imported-by filter options: owner-only, hidden while viewing-as
    # (entries.md B65 / permissions B28).
    if actor.allowed("can_transfer") and not actor.is_viewing_as:
        importer_rows = session.execute(
            select(User.id, User.idauth_username, User.email)
            .where(
                User.id.in_(
                    select(distinct(ImportBatch.imported_by)).where(
                        ImportBatch.tenant_id == actor.tenant_id,
                        ImportBatch.imported_by.is_not(None),
                    )
                )
            )
            .order_by(User.idauth_username)
        ).all()
        options["importers"] = [
            {"id": str(uid), "label": username or email}
            for uid, username, email in importer_rows
        ]
    else:
        options["importers"] = []
    return options