"""Location-admin <-> asset_admin role sync + region-permission helpers.

Single source of truth for two derived facts:

1. **Who holds the ``asset_admin`` role?**
   Fully auto-derived from ``asset_location_admins.person_id`` (plus
   the office-less ``asset_global_admin`` co-derivation). Every
   mutation of a location's admin list calls
   :func:`sync_asset_admin_role_for_person` on the affected people; the
   daily ``asset.asset_admin_role_reconcile`` cron is the backstop.

2. **Which locations does a given person administer?**
   :func:`get_location_ids_managed_by`, consumed by the region gate
   :func:`person_can_manage_asset` to decide whether a non-global admin
   may mutate a given asset.

Why locations and not offices (migration 535): ``asset_items.location``
is a managed column — every write goes through
:class:`ManagedLocationResolver`, and 100% of production assets carry a
valid managed value. Anchoring authority there makes "who covers this
asset" a direct lookup. The previous office-anchored model had to infer
it through a four-tier ladder (holder office → assignee city → an
``office.asset_location`` bridge column → the unique office in that
country), and each tier was a place for the answer to be wrong.

The office admin list still exists as
:class:`~app.domains.asset.models.AssetOfficeLocalAdmin` — the 当地
Admin, a local IT contact for person-anchored HR flows. It grants no
asset authority. See :mod:`app.domains.asset.office_local_admins`.
"""
from __future__ import annotations

import uuid
from collections.abc import Iterable

from sqlalchemy import delete, insert, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthRole, AuthUserRole
from app.domains.asset.location_resolver import LocationIndex
from app.domains.asset.models import (
    AssetItem,
    AssetLocation,
    AssetLocationAdmin,
)
from app.identity.person_models import IdentityPerson

# ── Role auto-derive ────────────────────────────────────────────────


async def _get_asset_admin_role_id(db: AsyncSession) -> uuid.UUID | None:
    """Resolve the ``asset_admin`` row id from ``auth_roles``."""
    row = await db.execute(
        select(AuthRole.id).where(AuthRole.name == "asset_admin")
    )
    return row.scalar_one_or_none()


async def _person_holds_asset_global_admin(
    db: AsyncSession, person_id: uuid.UUID,
) -> bool:
    """True when the person holds the manually-granted
    ``asset_global_admin`` role (全局资产管理员)."""
    row = await db.scalar(
        select(AuthUserRole.user_id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(
            AuthUserRole.user_id == person_id,
            AuthRole.name == "asset_global_admin",
        )
        .limit(1)
    )
    return row is not None


async def _person_should_hold_role(db: AsyncSession, person_id: uuid.UUID) -> bool:
    """A person holds ``asset_admin`` iff they are an active,
    non-deleted Person AND at least one grant source applies:

      * they appear in ``asset_location_admins`` for at least one
        location (the regional path), OR
      * they hold ``asset_global_admin`` (the location-less global
        path, 2026-06-11) — co-deriving ``asset_admin`` is what lets
        every existing entry gate / frontend surface admit global
        asset admins without enumerating a second role name.
    """
    person = await db.get(IdentityPerson, person_id)
    if person is None or person.deleted_at is not None:
        return False
    if (person.employment_status or "active") != "active":
        return False
    has_location = await db.scalar(
        select(AssetLocationAdmin.person_id)
        .where(AssetLocationAdmin.person_id == person_id)
        .limit(1)
    )
    if has_location is not None:
        return True
    return await _person_holds_asset_global_admin(db, person_id)


async def sync_asset_admin_role_for_person(
    db: AsyncSession, person_id: uuid.UUID,
) -> None:
    """Ensure ``auth_user_roles`` matches ``_person_should_hold_role``
    for the given person. Idempotent.

    Caller is responsible for committing the transaction — this
    function only stages the change.
    """
    role_id = await _get_asset_admin_role_id(db)
    if role_id is None:
        return  # role row missing; nothing to sync

    should_hold = await _person_should_hold_role(db, person_id)
    existing = await db.execute(
        select(AuthUserRole).where(
            AuthUserRole.user_id == person_id,
            AuthUserRole.role_id == role_id,
        )
    )
    existing_row = existing.scalar_one_or_none()

    if should_hold and existing_row is None:
        db.add(AuthUserRole(user_id=person_id, role_id=role_id))
    elif (not should_hold) and existing_row is not None:
        # ORM delete, not ``delete(AuthUserRole).where(...)``: the
        # access-cycle ledger listener observes the unit of work, and a
        # Core-level DELETE would silently strip the role without closing
        # its holding interval — leaving "still an asset admin" as the
        # permanent point-in-time answer. The row is already loaded, so
        # this costs no extra round-trip.
        await db.delete(existing_row)


async def reconcile_asset_admin_role(db: AsyncSession) -> dict[str, int]:
    """Full reconciliation: rebuild the asset_admin grant set from
    ``asset_location_admins``. Returns a counters dict.

    Called by the daily scheduled job
    ``asset.asset_admin_role_reconcile`` as a backstop in case the
    in-process sync ever drops a row (e.g. a code path that mutates
    the join table outside the canonical service). Cheap operation
    (~hundreds of rows at most), worth running every day.
    """
    role_id = await _get_asset_admin_role_id(db)
    if role_id is None:
        return {"role_missing": 1}

    # Persons who SHOULD hold the role.
    should = await db.execute(
        select(AssetLocationAdmin.person_id)
        .join(IdentityPerson, IdentityPerson.id == AssetLocationAdmin.person_id)
        .where(
            IdentityPerson.deleted_at.is_(None),
            (IdentityPerson.employment_status == "active")
            | (IdentityPerson.employment_status.is_(None)),
        )
        .distinct()
    )
    should_set: set[uuid.UUID] = {row[0] for row in should.all()}

    # Second grant source: active holders of ``asset_global_admin``
    # co-hold ``asset_admin`` (location-less global path). Without this
    # union the daily reconciler would revoke asset_admin from every
    # global asset admin who isn't also listed against a location.
    global_holders = await db.execute(
        select(AuthUserRole.user_id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .join(IdentityPerson, IdentityPerson.id == AuthUserRole.user_id)
        .where(
            AuthRole.name == "asset_global_admin",
            IdentityPerson.deleted_at.is_(None),
            (IdentityPerson.employment_status == "active")
            | (IdentityPerson.employment_status.is_(None)),
        )
        .distinct()
    )
    should_set |= {row[0] for row in global_holders.all()}

    # Persons who CURRENTLY hold the role.
    current = await db.execute(
        select(AuthUserRole.user_id).where(AuthUserRole.role_id == role_id)
    )
    current_set: set[uuid.UUID] = {row[0] for row in current.all()}

    to_grant = should_set - current_set
    to_revoke = current_set - should_set

    if to_revoke:
        # Load-then-ORM-delete so the access-cycle ledger listener closes
        # each holding interval. Bounded by the asset_admin holder count
        # (tens of rows), so the extra SELECT is immaterial next to
        # losing the point-in-time answer.
        stale_rows = (
            await db.execute(
                select(AuthUserRole).where(
                    AuthUserRole.role_id == role_id,
                    AuthUserRole.user_id.in_(to_revoke),
                )
            )
        ).scalars().all()
        for stale in stale_rows:
            await db.delete(stale)
    for pid in to_grant:
        db.add(AuthUserRole(user_id=pid, role_id=role_id))

    return {
        "granted": len(to_grant),
        "revoked": len(to_revoke),
        "total_expected": len(should_set),
    }


# ── Location-admin list management ──────────────────────────────────


async def replace_location_admins(
    db: AsyncSession,
    location_id: uuid.UUID,
    person_ids: Iterable[uuid.UUID],
    *,
    primary_person_id: uuid.UUID | None = None,
) -> list[uuid.UUID]:
    """Set the full admin list for a location to ``person_ids``.

    Returns the person_ids whose membership actually changed (added or
    removed) — the forensically useful set for the audit log, as
    opposed to the submitted list. Caller commits the transaction.

    ``primary_person_id`` defaults to the first entry. The primary is
    the single recipient for flows that address one responsible party
    (inventory office-held confirmations, recycle notifications).
    """
    person_id_list = [pid for pid in person_ids if pid is not None]
    # Dedup preserving order.
    seen: set[uuid.UUID] = set()
    deduped: list[uuid.UUID] = []
    for pid in person_id_list:
        if pid not in seen:
            seen.add(pid)
            deduped.append(pid)
    person_id_list = deduped

    # Validate every submitted person is an active, non-deleted Person.
    # Without this an admin could stuff terminated / departed UUIDs into
    # the join table — they wouldn't get the role (the sync filter
    # excludes them) but would pollute ``list_location_admins`` output
    # and could silently re-grant on rehire.
    if person_id_list:
        from app.core.exceptions import ValidationError  # noqa: PLC0415

        rows = await db.execute(
            select(IdentityPerson.id).where(
                IdentityPerson.id.in_(person_id_list),
                IdentityPerson.deleted_at.is_(None),
                (IdentityPerson.employment_status == "active")
                | (IdentityPerson.employment_status.is_(None)),
            )
        )
        valid_ids = {row[0] for row in rows.all()}
        invalid = [pid for pid in person_id_list if pid not in valid_ids]
        if invalid:
            raise ValidationError(
                "location admin candidates include inactive / "
                "departed / unknown persons: "
                + ", ".join(str(p) for p in invalid),
            )

    if primary_person_id is not None and primary_person_id not in person_id_list:
        from app.core.exceptions import ValidationError  # noqa: PLC0415

        raise ValidationError(
            "primary_person_id must be one of the submitted location admins",
        )
    if primary_person_id is None and person_id_list:
        primary_person_id = person_id_list[0]

    existing_rows = await db.execute(
        select(AssetLocationAdmin.person_id).where(
            AssetLocationAdmin.location_id == location_id,
        )
    )
    existing_set: set[uuid.UUID] = {row[0] for row in existing_rows.all()}
    new_set: set[uuid.UUID] = set(person_id_list)

    to_remove = existing_set - new_set
    to_add = new_set - existing_set

    if to_remove:
        await db.execute(
            delete(AssetLocationAdmin).where(
                AssetLocationAdmin.location_id == location_id,
                AssetLocationAdmin.person_id.in_(to_remove),
            )
        )
    for pid in to_add:
        await db.execute(
            insert(AssetLocationAdmin).values(
                location_id=location_id,
                person_id=pid,
                is_primary=(pid == primary_person_id),
            )
        )

    # Re-flag is_primary across the whole location, so a primary change
    # with an unchanged membership list still lands. Clearing first keeps
    # the "at most one primary" invariant even if a prior direct DB edit
    # set two.
    if new_set:
        await db.execute(
            update(AssetLocationAdmin)
            .where(AssetLocationAdmin.location_id == location_id)
            .values(is_primary=False)
        )
        await db.execute(
            update(AssetLocationAdmin)
            .where(
                AssetLocationAdmin.location_id == location_id,
                AssetLocationAdmin.person_id == primary_person_id,
            )
            .values(is_primary=True)
        )

    # Anyone whose membership changed needs a role re-sync.
    affected = to_remove | to_add
    for pid in affected:
        await sync_asset_admin_role_for_person(db, pid)

    return list(affected)


def _admin_row(person_id, is_primary, display_name, work_email) -> dict:
    return {
        "person_id": person_id,
        "person_name": display_name,
        "work_email": work_email,
        "is_primary": bool(is_primary),
    }


async def list_location_admins(
    db: AsyncSession, location_id: uuid.UUID,
) -> list[dict]:
    """Return ``[{person_id, person_name, work_email, is_primary}]``
    for the location, primary first then by display name.

    Departed people and soft-deleted rows are filtered out so the UI
    doesn't render ghost admins. The join row may survive a departure
    (the leaver guard sweeps it, but a sweep failure is non-fatal by
    design) — it is correctly hidden from operator-facing surfaces
    either way.
    """
    rows = await db.execute(
        select(
            AssetLocationAdmin.person_id,
            AssetLocationAdmin.is_primary,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
        )
        .join(IdentityPerson, IdentityPerson.id == AssetLocationAdmin.person_id)
        .where(
            AssetLocationAdmin.location_id == location_id,
            IdentityPerson.deleted_at.is_(None),
            (IdentityPerson.employment_status == "active")
            | (IdentityPerson.employment_status.is_(None)),
        )
        .order_by(AssetLocationAdmin.is_primary.desc(), IdentityPerson.display_name)
    )
    return [_admin_row(*r) for r in rows.all()]


async def list_location_admin_ids_bulk(
    db: AsyncSession, location_ids: Iterable[uuid.UUID],
) -> dict[uuid.UUID, list[dict]]:
    """Bulk variant for the list endpoint — one query for all locations."""
    location_id_list = list(location_ids)
    if not location_id_list:
        return {}
    rows = await db.execute(
        select(
            AssetLocationAdmin.location_id,
            AssetLocationAdmin.person_id,
            AssetLocationAdmin.is_primary,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
        )
        .join(IdentityPerson, IdentityPerson.id == AssetLocationAdmin.person_id)
        .where(
            AssetLocationAdmin.location_id.in_(location_id_list),
            IdentityPerson.deleted_at.is_(None),
            (IdentityPerson.employment_status == "active")
            | (IdentityPerson.employment_status.is_(None)),
        )
        .order_by(AssetLocationAdmin.is_primary.desc(), IdentityPerson.display_name)
    )
    out: dict[uuid.UUID, list[dict]] = {lid: [] for lid in location_id_list}
    for r in rows.all():
        out[r.location_id].append(
            _admin_row(r.person_id, r.is_primary, r.display_name, r.work_email)
        )
    return out


async def resolve_location_primary_admins_bulk(
    db: AsyncSession, location_ids: Iterable[uuid.UUID],
) -> dict[uuid.UUID, uuid.UUID | None]:
    """``{location_id: primary_admin_person_id | None}``.

    For flows that address exactly one responsible party. The rule:
    the row flagged ``is_primary``; failing that, the sole admin when
    there is exactly one. With several admins and no primary flag the
    answer is ``None`` — picking arbitrarily would send a
    "you are accountable for this" message to someone chosen by row
    order, and the callers all have an IT fallback that is honest
    about being a fallback.
    """
    by_location = await list_location_admin_ids_bulk(db, location_ids)
    out: dict[uuid.UUID, uuid.UUID | None] = {}
    for location_id, admins in by_location.items():
        primary = next((a["person_id"] for a in admins if a["is_primary"]), None)
        if primary is None and len(admins) == 1:
            primary = admins[0]["person_id"]
        out[location_id] = primary
    return out


# ── Region-permission helpers ───────────────────────────────────────


async def get_location_ids_managed_by(
    db: AsyncSession, person_id: uuid.UUID,
) -> set[uuid.UUID]:
    """Set of location IDs where this person is listed as an admin.

    Deliberately unfiltered by ``is_active``: an administrator keeps
    authority over the assets left behind at a location after an
    operator deactivates it. :func:`active_location_ids_managed_by` is
    the narrower variant for audience construction.
    """
    rows = await db.execute(
        select(AssetLocationAdmin.location_id).where(
            AssetLocationAdmin.person_id == person_id,
        )
    )
    return {row[0] for row in rows.all()}


async def active_location_ids_managed_by(
    db: AsyncSession, person_id: uuid.UUID,
) -> set[uuid.UUID]:
    """Active-only variant, for building intelligence audiences.

    A snapshot audience keyed on a retired location would keep
    producing a briefing about a place the company no longer operates.
    The write gate uses the unfiltered set instead — see
    :func:`get_location_ids_managed_by`.
    """
    rows = await db.execute(
        select(AssetLocationAdmin.location_id)
        .join(AssetLocation, AssetLocation.id == AssetLocationAdmin.location_id)
        .join(IdentityPerson, IdentityPerson.id == AssetLocationAdmin.person_id)
        .where(
            AssetLocationAdmin.person_id == person_id,
            AssetLocation.is_active.is_(True),
            IdentityPerson.deleted_at.is_(None),
            (IdentityPerson.employment_status == "active")
            | (IdentityPerson.employment_status.is_(None)),
        )
    )
    return {row[0] for row in rows.all()}


async def resolve_asset_location_ids_bulk(
    db: AsyncSession,
    assets: Iterable[AssetItem],
    *,
    index: LocationIndex | None = None,
) -> dict[uuid.UUID, uuid.UUID | None]:
    """``{asset_id: location_id | None}`` — the authoritative resolver.

    Two tiers, both keyed on managed locations:

    1. ``asset.location``, the managed column every write path routes
       through :class:`ManagedLocationResolver`. This is the answer for
       effectively every asset.
    2. The assignee's ``work_location_name`` / ``city``, for an asset
       whose own location was never set (or was set to a value the
       master no longer lists).

    ``None`` means "outside every regional administrator's scope" —
    only the global tier may act on it.

    Note what is *not* here: the office-held pointer. An office-held
    asset carries its managed location like any other, so tier 1
    answers for it too, and reading ``holder_office_id`` would
    reintroduce the office dependency this model exists to remove.
    """
    rows = list(assets)
    if not rows:
        return {}
    if index is None:
        index = await LocationIndex.load(db)

    person_ids = {row.assigned_to for row in rows if row.assigned_to is not None}
    person_locations: dict[uuid.UUID, tuple[str | None, str | None]] = {}
    if person_ids:
        for person_id, work_location, city in (
            await db.execute(
                select(
                    IdentityPerson.id,
                    IdentityPerson.work_location_name,
                    IdentityPerson.city,
                ).where(IdentityPerson.id.in_(person_ids))
            )
        ).all():
            person_locations[person_id] = (work_location, city)

    resolved: dict[uuid.UUID, uuid.UUID | None] = {}
    for asset in rows:
        location_id = index.resolve_id(asset.location)
        if location_id is None and asset.assigned_to is not None:
            location_id = index.resolve_first_id(
                *person_locations.get(asset.assigned_to, (None, None))
            )
        resolved[asset.id] = location_id
    return resolved


async def get_asset_location_id(
    db: AsyncSession, asset: AssetItem,
) -> uuid.UUID | None:
    """Single-asset form of :func:`resolve_asset_location_ids_bulk`."""
    return (await resolve_asset_location_ids_bulk(db, [asset]))[asset.id]


async def resolve_person_location_ids_bulk(
    db: AsyncSession,
    person_ids: Iterable[uuid.UUID],
    *,
    index: LocationIndex | None = None,
) -> dict[uuid.UUID, uuid.UUID | None]:
    """``{person_id: location_id | None}`` from the HRIS work location.

    ``work_location_name`` first, then ``city`` — the same order and
    the same resolver the asset side uses, so a person and the assets
    they hold always land in the same region.
    """
    ids = set(person_ids)
    if not ids:
        return {}
    if index is None:
        index = await LocationIndex.load(db)
    people = (
        await db.execute(
            select(
                IdentityPerson.id,
                IdentityPerson.work_location_name,
                IdentityPerson.city,
            ).where(IdentityPerson.id.in_(ids))
        )
    ).all()
    result: dict[uuid.UUID, uuid.UUID | None] = {pid: None for pid in ids}
    for person_id, work_location, city in people:
        result[person_id] = index.resolve_first_id(work_location, city)
    return result


async def get_person_location_id(
    db: AsyncSession, person_id: uuid.UUID,
) -> uuid.UUID | None:
    """Single-person form of :func:`resolve_person_location_ids_bulk`."""
    return (await resolve_person_location_ids_bulk(db, [person_id])).get(person_id)


async def person_can_manage_asset(
    db: AsyncSession, person_id: uuid.UUID, asset: AssetItem,
    *, role_names: Iterable[str] = (),
) -> bool:
    """Region-scoped permission gate for asset write operations.

    Authorization rules, in priority order:

    1. ``global_admin`` / ``asset_global_admin`` always pass —
       :func:`app.core.rbac.has_global_asset_scope` (orphan assets
       with no resolvable location included).
    2. For a regional ``asset_admin``: the asset's resolved location
       must be one they administer. An asset whose location resolves
       to nothing belongs to no region, so only the global tier can
       act on it.
    3. Other roles fail (the route-level dep already rejects them).
    """
    from app.core.rbac import has_global_asset_scope  # noqa: PLC0415

    if has_global_asset_scope(role_names):
        return True

    asset_location = await get_asset_location_id(db, asset)
    if asset_location is None:
        return False

    managed = await get_location_ids_managed_by(db, person_id)
    return asset_location in managed