"""The shared entries filter contract (rpcs-and-authz §1) as ONE
composable Select builder, plus THE ONE canonical status predicate
(§1.1) — the four divergent legacy copies do not port (02 §4).

FK-based display (rebuild delta #4): a resolved row's operator/client/
project come from the entity chain (projects -> parties); an unassigned
row (project_id IS NULL) shows its immutable raw_* triple. All §1 text
filters run against those display expressions, preserving the exact
legacy semantics: ILIKE **contains**, raw pattern concatenation (`%`/`_`
in user input act as wildcards), and NULL columns never match.
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass
from datetime import datetime

from sqlalchemy import Select, and_, case, false, or_, select, true
from sqlalchemy.orm import aliased

from app.models import ImportBatch, InvoiceLine, Party, Project, TimeEntry

#: Blocked-cell message order — first missing field wins (entries.md
#: B22, adapted to the FK model; start_at is NOT NULL by schema).
ELIGIBILITY_MESSAGE_ORDER: tuple[tuple[str, str], ...] = (
    ("worker", "Unknown person — pick a Team member."),
    ("operator", "Missing operator (BW / PlusROI)."),
    ("client", "Missing client."),
    ("project", "Missing project."),
    ("description", "Missing description."),
    ("end_at", "Missing end time."),
    ("source_user_email", "Missing source email."),
)


class EntryJoins:
    """The entries join graph + display expressions, built once and
    shared by every reporting query (list, totals, options, summaries).
    Instantiate per query — aliases must not leak across statements."""

    def __init__(self) -> None:
        self.project = aliased(Project, name="p")
        self.operator_party = aliased(Party, name="op")
        self.client_party = aliased(Party, name="cp")
        self.worker_party = aliased(Party, name="wp")

        # Unlabeled expressions — callers .label() them at select time
        # (labels don't belong in WHERE/GROUP BY clauses).
        unassigned = TimeEntry.project_id.is_(None)
        self.operator_expr = case(
            (unassigned, TimeEntry.raw_operator), else_=self.operator_party.name
        )
        self.client_expr = case(
            (unassigned, TimeEntry.raw_client), else_=self.client_party.name
        )
        self.project_expr = case(
            (unassigned, TimeEntry.raw_project), else_=self.project.name
        )
        self.user_expr = self.worker_party.name

    def select_from(self, *columns) -> Select:
        return (
            select(*columns)
            .select_from(TimeEntry)
            .outerjoin(self.project, self.project.id == TimeEntry.project_id)
            .outerjoin(
                self.operator_party,
                self.operator_party.id == self.project.operator_party_id,
            )
            .outerjoin(
                self.client_party, self.client_party.id == self.project.client_party_id
            )
            .outerjoin(
                self.worker_party, self.worker_party.id == TimeEntry.worker_party_id
            )
        )


#: The ONE eligibility conjunction (ELIGIBILITY_COLUMNS ported to FKs:
#: worker_party_id replaces team_member_id; project_id subsumes the
#: operator/client/project text triple — entries.md rebuild delta).
def _eligible():
    return and_(
        TimeEntry.worker_party_id.is_not(None),
        TimeEntry.project_id.is_not(None),
        TimeEntry.description.is_not(None),
        TimeEntry.end_at.is_not(None),
        TimeEntry.source_user_email.is_not(None),
    )


def _broken():
    return or_(
        TimeEntry.worker_party_id.is_(None),
        TimeEntry.project_id.is_(None),
        TimeEntry.description.is_(None),
        TimeEntry.end_at.is_(None),
        TimeEntry.source_user_email.is_(None),
    )


def _locked():
    return TimeEntry.invoice_line_id.is_not(None)


def status_condition(status: str | None):
    """§1.1 canonical predicate, invoice-line lock edition. Every row is
    in exactly one of {pending, blocked, applied}; 'transferred' is the
    back-compat alias of 'applied'; unknown values match nothing (no
    error — parity)."""
    if status in (None, "", "all"):
        return true()
    if status == "pending":
        return and_(~_locked(), _eligible())
    if status in ("applied", "transferred"):
        return _locked()
    if status == "blocked":
        return and_(~_locked(), _broken())
    return false()


def status_case():
    """Per-row status label built from the SAME condition objects — the
    predicate has one implementation, used by filter and display alike."""
    return case(
        (_locked(), "applied"),
        (_eligible(), "pending"),
        else_="blocked",
    ).label("status")


def status_sort_rank():
    """Numeric rank for ORDER BY status, from the SAME _locked()/_eligible()
    predicates as status_case (applied=2 > pending=1 > blocked=0). The
    sort key therefore holds NO second copy of the eligibility
    conjunction — one predicate, everywhere (T-PHX-009)."""
    return case((_locked(), 2), (_eligible(), 1), else_=0)


def blocked_reason(
    *,
    worker: str | None,
    operator: str | None,
    client: str | None,
    project: str | None,
    description: str | None,
    end_at,
    source_user_email: str | None,
) -> str | None:
    """Display-layer message for a blocked row (B22 order, first missing
    field wins). Presentation over the SAME fields the predicate uses."""
    values = {
        "worker": worker,
        "operator": operator,
        "client": client,
        "project": project,
        "description": description,
        "end_at": end_at,
        "source_user_email": source_user_email,
    }
    for key, message in ELIGIBILITY_MESSAGE_ORDER:
        if values[key] is None:
            return message
    return None


@dataclass(frozen=True)
class EntryFilters:
    """§1 params. Every field None/'all' = filter off."""

    start: datetime | None = None  # start_at >= start
    end: datetime | None = None  # start_at <  end   (half-open)
    status: str | None = None
    batch: uuid.UUID | None = None
    q: str | None = None
    project: str | None = None
    user: str | None = None
    source_email: str | None = None
    client: str | None = None
    operator: str | None = None
    imported_by: uuid.UUID | None = None
    missing_info: bool = False
    invoice: uuid.UUID | None = None  # rebuild: applied to list AND totals


def _contains(column, value: str):
    # Raw concatenation on purpose: `%`/`_` in user input act as
    # wildcards, exactly like the legacy RPCs (§1 semantics).
    return column.ilike(f"%{value}%")


def apply_filters(stmt: Select, joins: EntryJoins, f: EntryFilters) -> Select:
    """AND the whole §1 filter set onto a Select built from
    joins.select_from(...). Tenant/visibility scoping is scope()'s job,
    NOT this function's — callers compose scope(actor, ...) around it."""
    conditions = [status_condition(f.status)]
    if f.start is not None:
        conditions.append(TimeEntry.start_at >= f.start)
    if f.end is not None:
        conditions.append(TimeEntry.start_at < f.end)
    if f.batch is not None:
        conditions.append(TimeEntry.import_id == f.batch)
    if f.q:
        pat = f"%{f.q}%"
        # exactly these 4 columns — NOT operator, NOT worker (§1 p_q)
        conditions.append(
            or_(
                joins.project_expr.ilike(pat),
                TimeEntry.description.ilike(pat),
                TimeEntry.source_user_email.ilike(pat),
                joins.client_expr.ilike(pat),
            )
        )
    if f.project:
        conditions.append(_contains(joins.project_expr, f.project))
    if f.user:
        conditions.append(_contains(joins.user_expr, f.user))
    if f.source_email:
        conditions.append(_contains(TimeEntry.source_user_email, f.source_email))
    if f.client:
        conditions.append(_contains(joins.client_expr, f.client))
    if f.operator:
        conditions.append(_contains(joins.operator_expr, f.operator))
    if f.imported_by is not None:
        conditions.append(
            TimeEntry.import_id.in_(
                select(ImportBatch.id).where(ImportBatch.imported_by == f.imported_by)
            )
        )
    if f.missing_info:
        conditions.append(
            or_(
                joins.operator_expr.is_(None),
                joins.client_expr.is_(None),
                joins.project_expr.is_(None),
            )
        )
    if f.invoice is not None:
        conditions.append(
            TimeEntry.invoice_line_id.in_(
                select(InvoiceLine.id).where(InvoiceLine.invoice_id == f.invoice)
            )
        )
    return stmt.where(*conditions)