"""Read-only summary powering the "必须转让" tally on
``/admin/offboarding``.

Owner reported 2026-05-07: the existing overview page only counted
``asset_software_catalog`` rows whose ``offboarding_policy`` equals
``must_transfer``, so the tally read **0** even though hundreds of
cloud subscriptions / email groups / IAM users / physical assets
were owned by employees and would need transfer at offboarding time.

This service computes the breadth of the "must transfer" universe
by category. Categories are deliberately disjoint — each row in the
union belongs to exactly one bucket — so the totals add up to the
real count of transferable ownership relations.

The page calling us is admin-only (``require_admin``); we don't
re-gate here. Each query is a single ``SELECT COUNT(*)`` aggregate;
no per-row Python filtering, no joins to ``identity_persons`` since
the count we want is "things with an owner FK set", not "things
whose owner happens to be active right now" — that conflation would
hide the very rows admins need to plan transfer policy around.
"""
from __future__ import annotations

from dataclasses import dataclass
from typing import Any

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


@dataclass(frozen=True, slots=True)
class TransferableCategoryCount:
    """One row in the summary response.

    ``key`` is a stable enum string the frontend keys translations
    + icons off; ``count`` is the integer aggregate; ``admin_path``
    is the canonical admin page admins land on when they click the
    card.
    """

    key: str
    count: int
    admin_path: str


@dataclass(frozen=True, slots=True)
class TransferableSummary:
    categories: list[TransferableCategoryCount]

    @property
    def total(self) -> int:
        return sum(c.count for c in self.categories)


class TransferableSummaryService:
    """Aggregates the six "must transfer at offboarding" categories.

    Cross-domain reads via inline imports — same pattern
    ``OffboardingService`` uses for its blocker scan
    (``server/app/domains/offboarding/service.py``).
    """

    def __init__(self, db: AsyncSession) -> None:
        self.db = db

    async def get_summary(self) -> TransferableSummary:
        return TransferableSummary(categories=[
            TransferableCategoryCount(
                key="software",
                count=await self._software_count(),
                admin_path="/admin/assets/software-catalog",
            ),
            TransferableCategoryCount(
                key="cloud_scope",
                count=await self._cloud_scope_count(),
                admin_path="/admin/cloud/scopes",
            ),
            TransferableCategoryCount(
                key="domain",
                count=await self._domain_count(),
                admin_path="/admin/cloud/domains",
            ),
            TransferableCategoryCount(
                key="email_group",
                count=await self._email_group_count(),
                admin_path="/company/mail-groups",
            ),
            TransferableCategoryCount(
                key="asset_item",
                count=await self._asset_item_count(),
                admin_path="/admin/assets/inventory",
            ),
            TransferableCategoryCount(
                key="iam_user",
                count=await self._iam_user_count(),
                admin_path="/admin/cloud/iam-users",
            ),
        ])

    # ── per-category counters ─────────────────────────────────────

    async def _software_count(self) -> int:
        """Software entries explicitly configured with the
        ``must_transfer`` offboarding policy. Active rows only —
        ``status='active'`` is the catalog's enabled flag; archived
        software should not contribute to the live tally."""
        from app.domains.asset.models import AssetSoftwareCatalog

        return int(await self.db.scalar(
            select(func.count(AssetSoftwareCatalog.id)).where(
                AssetSoftwareCatalog.offboarding_policy == "must_transfer",
                AssetSoftwareCatalog.status == "active",
            )
        ) or 0)

    async def _cloud_scope_count(self) -> int:
        """Cloud subscriptions (Aliyun / AWS / Azure / Tencent / …)
        currently bound to an owner. Owner-transfer is the universal
        policy at offboarding (no per-scope enum), so anything with
        ``owner_person_id IS NOT NULL`` and not deleted/decommissioned
        counts."""
        from app.domains.cmdb.models import CmdbCloudScope

        return int(await self.db.scalar(
            select(func.count(CmdbCloudScope.id)).where(
                CmdbCloudScope.owner_person_id.is_not(None),
                CmdbCloudScope.status == "active",
                CmdbCloudScope.deleted_at.is_(None),
            )
        ) or 0)

    async def _domain_count(self) -> int:
        """Two domain ownership streams sum into one card:

        * ``domain_registrations`` — self-service sub-domain allocations
          (``person_id`` is the assignee). Counts ``status='approved'``
          rows — the offboarding scanner uses the same filter
          (``DOMAIN_REG_STATUS_APPROVED``) so the must-transfer tally
          stays in lockstep with what actually generates a leaver
          blocker. Pending / cancelled / revoked rows aren't held.
        * ``cmdb_domains`` — registrar-discovered DNS records
          (``owner_person_id`` is the admin-assigned owner).

        They live in different domains and have different admin pages,
        but to the offboarding-policy reader they're one concept:
        "domains tied to a person who'll need to hand them off".
        """
        from app.domains.cmdb.models import CmdbDomain
        from app.domains.domain_registration.models import (
            DOMAIN_REG_STATUS_APPROVED,
            DomainRegistration,
        )

        registrations = int(await self.db.scalar(
            select(func.count(DomainRegistration.id)).where(
                DomainRegistration.person_id.is_not(None),
                DomainRegistration.status == DOMAIN_REG_STATUS_APPROVED,
            )
        ) or 0)
        cloud_domains = int(await self.db.scalar(
            select(func.count(CmdbDomain.id)).where(
                CmdbDomain.owner_person_id.is_not(None),
                CmdbDomain.deleted_at.is_(None),
            )
        ) or 0)
        return registrations + cloud_domains

    async def _email_group_count(self) -> int:
        """Distinct ACTIVE email groups that carry at least one owner.

        Counted by ``email_group_id`` (not by row) because a single
        group can have multiple co-owners — each leaver among them
        still triggers a hand-off, but the *group* is the
        transferable unit the admin reasons about.

        Filtering ``EmailGroup.status='active'`` mirrors the
        offboarding scanner / handover lookup
        (``server/app/domains/offboarding/service.py:2340``); a
        deleted-or-archived group can't be reassigned, so it doesn't
        belong in the must-transfer tally.
        """
        from app.domains.email_group.models import EmailGroup, EmailGroupOwner

        return int(await self.db.scalar(
            select(func.count(func.distinct(EmailGroupOwner.email_group_id)))
            .select_from(EmailGroupOwner)
            .join(EmailGroup, EmailGroup.id == EmailGroupOwner.email_group_id)
            .where(EmailGroup.status == "active")
        ) or 0)

    async def _asset_item_count(self) -> int:
        """Physical asset items currently held (assigned or borrowed).

        Reuses ``ASSET_IN_USE_STATUSES`` so this stays in lockstep
        with the offboarding scanner's blocker logic — if the scanner
        flags a status as transfer-relevant, the overview counts it.
        """
        from app.domains.asset.models import ASSET_IN_USE_STATUSES, AssetItem

        return int(await self.db.scalar(
            select(func.count(AssetItem.id)).where(
                AssetItem.deleted_at.is_(None),
                AssetItem.status.in_(ASSET_IN_USE_STATUSES),
                # Either the persistent assignee OR a current borrower
                # ties the row to a person who'd need to hand it off.
                (AssetItem.assigned_to.is_not(None))
                | (AssetItem.borrowed_by.is_not(None)),
            )
        ) or 0)

    async def _iam_user_count(self) -> int:
        """Cloud IAM users (RAM accounts, AWS IAM, Azure AD app users,
        …) the platform has imported and that an admin has assigned
        an owner to. ``owner_person_id IS NULL`` rows are unowned
        leftovers that the IAM cleanup flow handles separately —
        they don't belong in the must-transfer tally."""
        from app.domains.cmdb.models import CmdbIamUser

        return int(await self.db.scalar(
            select(func.count(CmdbIamUser.id)).where(
                CmdbIamUser.owner_person_id.is_not(None),
                CmdbIamUser.deleted_at.is_(None),
            )
        ) or 0)


def serialize(summary: TransferableSummary) -> dict[str, Any]:
    """Wire shape for the admin overview frontend."""
    return {
        "categories": [
            {
                "key": c.key,
                "count": c.count,
                "admin_path": c.admin_path,
            }
            for c in summary.categories
        ],
        "total": summary.total,
    }