"""Department-subtree resolution — shared org-scope primitives.

A single source of truth for "what does filtering by department *X* mean".
Helix's department filters used to be split-brained: the software-cost page
rolled a chosen node's whole subtree up (pick *Business Operations* → also
get *IT Services & Security*), while the Users-list and Cloud-overview
filters matched the exact leaf only. Once every department picker became the
hierarchical org TreeSelect (2026-06-03), picking a parent node has to mean
"this department and everything under it" everywhere, or the parent rows in
the tree would silently return nothing.

The org tree is stored as a self-referential adjacency list on
``identity_org_nodes`` (``parent_org_node_source_id``) plus a denormalised
breadcrumb ``org_path`` of the form ``"PingCAP / Business Operations / IT
Services & Security"``. ``identity_org_memberships`` carries the same
``org_node_name`` / ``org_path`` per person. We match on the breadcrumb so a
descendant at any depth is covered without a recursive CTE, and the
``"/ {name}"`` separator guards against prefix-similar siblings
(*Security* vs *Security & Compliance*).

Department names are not globally unique — the same name can sit under two
BUs (``get_departments`` already dedups by name). Matching by name therefore
unions every same-named subtree, which is the platform-wide contract every
``*_department`` name column already follows.
"""

from __future__ import annotations

from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.sql.elements import ColumnElement

from app.identity.person_models import IdentityOrgMembership, IdentityOrgNode


def _subtree_path_clauses(
    name_col: ColumnElement[str],
    path_col: ColumnElement[str | None],
    department: str,
) -> ColumnElement[bool]:
    """OR predicate matching ``department`` and every descendant of it.

    ``org_path`` segments are joined by ``" / "`` (space-slash-space), e.g.
    ``"PingCAP / Security & Compliance / TeamX"``. Every descendant arm is
    anchored on that FULL separator so a prefix-similar sibling can never be
    pulled in — filtering by *Security* must NOT match *Security & Compliance*.
    The arms cover, respectively: an exact node-name hit, a single-segment
    path, ``dept`` as the root segment, ``dept`` as an interior segment, and
    ``dept`` as the deepest (leaf) segment. LIKE wildcards in ``dept`` are
    escaped so a name containing ``%`` / ``_`` stays literal.
    """
    dep = department.strip()
    esc = dep.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
    return or_(
        name_col == dep,                              # exact node-name hit
        path_col == dep,                              # single-segment path
        path_col.like(f"{esc} / %", escape="\\"),     # dep is the root segment
        path_col.like(f"% / {esc} / %", escape="\\"),  # dep is interior
        path_col.like(f"% / {esc}", escape="\\"),      # dep is the leaf
    )


def department_membership_subtree_clause(department: str) -> ColumnElement[bool]:
    """Subtree predicate over ``IdentityOrgMembership`` for a *primary*
    org-membership join. Caller is responsible for joining the membership in
    with ``is_primary`` / ``deleted_at`` guards — this only supplies the
    department breadcrumb match.
    """
    return _subtree_path_clauses(
        IdentityOrgMembership.org_node_name,
        IdentityOrgMembership.org_path,
        department,
    )


async def resolve_department_subtree_names(
    db: AsyncSession, department: str | None,
) -> set[str] | None:
    """Resolve a department name to the set of leaf ``org_node_name`` strings
    in its subtree (inclusive).

    Returns ``None`` when no filter is requested (caller skips filtering) and
    a ``set[str]`` otherwise — used to narrow denormalised leaf-name columns
    like ``cmdb_cost_facts.cost_owner_department`` with ``col.in_(names)`` so
    a cost ranking agrees with the headcount filter on what a parent
    department contains. The chosen name is always included even if no node
    matches, so an unknown name degrades to an exact match rather than
    widening scope.
    """
    if not department:
        return None
    dep = department.strip()
    rows = (
        await db.execute(
            select(IdentityOrgNode.org_node_name).where(
                IdentityOrgNode.org_node_type == "department",
                IdentityOrgNode.deleted_at.is_(None),
                _subtree_path_clauses(
                    IdentityOrgNode.org_node_name,
                    IdentityOrgNode.org_path,
                    dep,
                ),
            )
        )
    ).scalars().all()
    names = {r for r in rows if r}
    names.add(dep)
    return names