"""What does this person actually hold? — the mover's read-only scan.

Leaver and Mover are the same question asked with different stakes.
When somebody leaves, Helix enumerates twenty-odd kinds of thing they
hold and refuses to close the process until each one is revoked or
handed over. When somebody *moves*, Helix used to look at exactly one:
their cloud access grants. Everything else — the software seats their
old department paid for, the DNS records they own, the systems that
name them as the person to call, the API keys, the SSO grants, the
mailing-list ownerships — kept pointing at a person whose job is now
somebody else's job.

This module answers the same question the offboarding scanner answers,
with three deliberate differences:

* **It writes nothing.** ``OffboardingService._rescan`` creates
  ``OffboardingBlocker`` rows bound to a process; a role change has no
  process and needs no blockers. This is a pure read.
* **Nothing is auto-actioned.** Issue #623's standing policy is
  human-in-the-loop on org moves — legal asked for it after auto-revoke
  hit the wrong person mid-quarter. Every holding here becomes a
  question, never an action.
* **Some holdings are context, not questions.** Hardware follows the
  employee through a move (also #623). It is still listed, because a
  new manager reading a handover packet with no laptop on it will
  wonder where the laptop went — but it carries ``INFORMATIONAL`` and
  asks for no decision.

**Divergence is the risk this module carries.** Two enumerations of
"what a person holds" that drift apart mean a resource type gets added
to offboarding and silently never appears in a mover review — the exact
class of bug PR #3315 just spent a day fixing on the other side of the
same feature. ``RESOURCE_DISPOSITIONS`` below is keyed by the SAME
constants the offboarding scanner uses, and
``test_mover_holdings_cover_every_offboarding_resource_type`` fails the
build when a new one appears without a mover disposition. Adding a
resource type is therefore a two-line decision, not a silent omission.
"""
from __future__ import annotations

import logging
from dataclasses import dataclass, field
from typing import Any
from uuid import UUID

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

from app.domains.offboarding.models import (
    RES_AI_TOOL_APPLICATION,
    RES_API_KEY,
    RES_APPROVAL_FLOW_OWNER,
    RES_CDN_DOMAIN,
    RES_CERTIFICATE,
    RES_CLOUD_ACCESS_GRANT,
    RES_CLOUD_IAM_USER,
    RES_CLOUD_SCOPE_OWNER,
    RES_CLOUD_SERVER,
    RES_DATA_ACCESS_GRANT,
    RES_DNS_RECORD,
    RES_DOMAIN,
    RES_DOMAIN_REGISTRATION,
    RES_EMAIL_GROUP,
    RES_EMAIL_GROUP_MEMBER,
    RES_GITHUB_ORG_MEMBERSHIP,
    RES_GOOGLE_ORPHAN_OWNER,
    RES_HARDWARE,
    RES_IDENTITY_CONSUMER_OWNER,
    RES_LOAD_BALANCER,
    RES_OBJECT_STORAGE,
    RES_SOFTWARE_ACCOUNT,
    RES_SOFTWARE_IT_CONTACT,
    RES_SOFTWARE_OWNER,
    RES_SOFTWARE_PERMISSION,
    RES_SOFTWARE_SEAT,
    RES_SSO_GRANT,
    RES_SYSTEM_OWNER,
    RES_VPN_APPLICATION,
)

logger = logging.getLogger(__name__)

# ── Dispositions ────────────────────────────────────────────────────
#
# What a mover review ASKS about a holding. Never what it does — the
# platform does nothing to a mover's access without a human pressing a
# button (issue #623, not re-litigated here).

#: Needs a keep / revoke decision. The holding grants access or names
#: accountability, and both are department-shaped: the reason the
#: person had it may have moved with their old job.
REVIEW = "review"

#: Shown for context, no decision. The holding follows the person
#: through a move by policy, and a packet that omits it reads as
#: incomplete rather than as settled.
INFORMATIONAL = "informational"

#: Every offboarding resource type, mapped to what a MOVE should ask
#: about it. Keyed on the offboarding constants on purpose — see the
#: module docstring on divergence.
RESOURCE_DISPOSITIONS: dict[str, str] = {
    # Access the person holds. A move can invalidate the reason for all
    # of these, and none of them lapse on their own.
    RES_SOFTWARE_SEAT: REVIEW,
    RES_SOFTWARE_PERMISSION: REVIEW,
    RES_SOFTWARE_ACCOUNT: REVIEW,
    RES_API_KEY: REVIEW,
    RES_CLOUD_ACCESS_GRANT: REVIEW,
    RES_SSO_GRANT: REVIEW,
    RES_VPN_APPLICATION: REVIEW,
    RES_AI_TOOL_APPLICATION: REVIEW,
    RES_EMAIL_GROUP: REVIEW,
    # Write access to the company's repositories, and department-shaped: an
    # engineer who moves to a non-engineering team keeps it until somebody
    # looks. It also sits OUTSIDE the SSO cascade — nothing about a move
    # touches it on its own, which is exactly the case a review packet is for.
    RES_GITHUB_ORG_MEMBERSHIP: REVIEW,
    # A personal BigQuery dataset grant was approved for the job the person
    # HAD; a move is exactly when the dataset owner should look again.
    RES_DATA_ACCESS_GRANT: REVIEW,
    # Accountability the person carries. Nothing to revoke — somebody
    # has to still be the answer to "who owns this", and after a move
    # the honest answer may be someone else.
    RES_DOMAIN: REVIEW,
    RES_DNS_RECORD: REVIEW,
    RES_CLOUD_SERVER: REVIEW,
    RES_LOAD_BALANCER: REVIEW,
    RES_CDN_DOMAIN: REVIEW,
    RES_CLOUD_SCOPE_OWNER: REVIEW,
    RES_CLOUD_IAM_USER: REVIEW,
    RES_DOMAIN_REGISTRATION: REVIEW,
    RES_SOFTWARE_OWNER: REVIEW,
    RES_SOFTWARE_IT_CONTACT: REVIEW,
    RES_SYSTEM_OWNER: REVIEW,
    RES_APPROVAL_FLOW_OWNER: REVIEW,
    RES_GOOGLE_ORPHAN_OWNER: REVIEW,
    RES_IDENTITY_CONSUMER_OWNER: REVIEW,
    # 2026-09-05 (PR3). Buckets and manually-assigned certificate owners
    # are the same accountability shape as the network assets above.
    RES_OBJECT_STORAGE: REVIEW,
    RES_CERTIFICATE: REVIEW,
    # Mail-group membership survives a move on its own — an engineer who
    # moves off the finance team keeps reading finance-team@ until
    # somebody looks. Ownership (RES_EMAIL_GROUP) is a separate question.
    RES_EMAIL_GROUP_MEMBER: REVIEW,
    # Follows the employee. #623: assets stay with the person through a
    # role move; listed so the packet is complete, never asked about.
    RES_HARDWARE: INFORMATIONAL,
}


@dataclass(slots=True)
class Holding:
    """One thing the person holds, as the review packet sees it."""

    resource_type: str
    resource_id: UUID
    label: str
    disposition: str
    #: Free-form, resource-shaped context the card and the fact package
    #: render (plan name, scope external id, role key, …). Never used as
    #: an identifier — ``(resource_type, resource_id)`` is the key.
    detail: dict[str, Any] = field(default_factory=dict)


@dataclass(slots=True)
class HoldingsScan:
    holdings: list[Holding] = field(default_factory=list)
    #: Resource types whose query raised. A mover packet that silently
    #: dropped a whole category would be worse than one that says so:
    #: the reviewer would read a short list as "not much to decide".
    failed_types: list[str] = field(default_factory=list)

    @property
    def reviewable(self) -> list[Holding]:
        return [h for h in self.holdings if h.disposition == REVIEW]


async def scan_person_holdings(
    db: AsyncSession, person_id: UUID,
) -> HoldingsScan:
    """Enumerate everything ``person_id`` holds, for a mover review.

    Read-only. Each resource type is isolated: one domain's query
    failing (a table missing on an install that never enabled the
    domain, a vendor id column that is a string on one side and a UUID
    on the other) records the type in ``failed_types`` and lets the rest
    of the scan complete. The packet then shows the gap rather than
    silently under-reporting.
    """
    scan = HoldingsScan()
    for resource_type, loader in _LOADERS.items():
        try:
            found = await loader(db, person_id)
        except Exception:  # noqa: BLE001 — one domain must not end the scan
            logger.exception(
                "mover_holdings: %s scan failed for person=%s",
                resource_type, person_id,
            )
            scan.failed_types.append(resource_type)
            continue
        scan.holdings.extend(found)
    return scan


def _holding(
    resource_type: str,
    resource_id: UUID,
    label: str,
    **detail: Any,
) -> Holding:
    return Holding(
        resource_type=resource_type,
        resource_id=resource_id,
        label=label,
        disposition=RESOURCE_DISPOSITIONS[resource_type],
        detail={k: v for k, v in detail.items() if v is not None},
    )


# ── Per-resource loaders ────────────────────────────────────────────
#
# Each mirrors the predicate the offboarding scanner uses for the same
# resource type. Where the two could reasonably differ, the comment says
# which one this is and why.


# ── Delta-relevant detail ───────────────────────────────────────────
#
# Three keys the delta engine (``mover_delta``) reads off ``detail``,
# added 2026-09-05. They are populated ONLY where the source table
# genuinely carries them; a loader that cannot answer leaves the key
# absent and the engine reports ``unknown`` rather than guessing.
#
#   ``policy_target`` — ``"<target_type>:<uuid>"``, matching the shape
#       ``provision_policies`` uses. Present only for holdings a policy
#       could possibly be written about (seats, permissions). Its
#       absence is what stops the engine from claiming that a DNS record
#       is "not in the policy set" — no policy can be written about a
#       DNS record, so policies are silent on it, and silence is not a
#       verdict.
#   ``granted_at`` — ISO instant the access began, for the provenance
#       signal ("granted while they sat in the department they just
#       left"). Ownership rows have no such moment and omit it.
#   ``peer_key`` — a stable identity for "the same thing" across people,
#       so the share of the mover's new peers holding it can be counted.
#       The label is NOT usable for this: two plans can print the same
#       name, and a rename would silently re-bucket everybody.


async def _hardware(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from sqlalchemy import or_ as sa_or

    from app.domains.asset.models import ASSET_IN_USE_STATUSES, AssetItem

    # ``borrowed_by`` as well as ``assigned_to`` (2026-09-05, PR3): a
    # borrowed laptop (status='borrowed', assigned_to=someone else) is
    # physically in this person's hands, which is the only thing either
    # consumer of this loader cares about. The T+1 residual scan had this
    # OR since the borrowed-asset blind spot; the mover packet did not,
    # so a mover could hand back nothing and still hold a device.
    rows = (await db.execute(
        select(AssetItem).where(
            sa_or(
                AssetItem.assigned_to == person_id,
                AssetItem.borrowed_by == person_id,
            ),
            AssetItem.status.in_(tuple(ASSET_IN_USE_STATUSES)),
        )
    )).scalars().all()
    return [
        _holding(
            RES_HARDWARE, item.id,
            item.asset_tag or str(item.id),
            model=item.model,
            status=item.status,
        )
        for item in rows
    ]


async def _software_seats(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.core.coordinators.jamf_seat_coordinator import (
        mdm_managed_software_ids,
    )
    from app.domains.asset.models import (
        AssetSoftwareCatalog,
        SoftwareLicensePlan,
        SoftwareSeatAllocation,
    )

    rows = (await db.execute(
        select(SoftwareSeatAllocation, SoftwareLicensePlan, AssetSoftwareCatalog)
        .join(
            SoftwareLicensePlan,
            SoftwareLicensePlan.id == SoftwareSeatAllocation.plan_id,
        )
        .join(
            AssetSoftwareCatalog,
            AssetSoftwareCatalog.id == SoftwareLicensePlan.software_id,
        )
        .where(
            SoftwareSeatAllocation.user_id == person_id,
            SoftwareSeatAllocation.status == "active",
        )
    )).all()
    # MDM enrolment follows the DEVICE, not the employee (migration 558).
    # Resolved through the SAME coordinator the offboarding scanner uses
    # rather than a local ``kind == 'mdm'`` guess — that column does not
    # exist, and a private copy of the rule would drift.
    mdm_ids = await mdm_managed_software_ids(db)
    out: list[Holding] = []
    for seat, plan, sw in rows:
        if sw.id in mdm_ids:
            continue
        out.append(_holding(
            RES_SOFTWARE_SEAT, seat.id,
            f"{sw.name} — {plan.name}",
            software=sw.name,
            plan=plan.name,
            policy_target=f"seat:{plan.id}",
            peer_key=f"seat:{plan.id}",
            granted_at=(
                seat.created_at.isoformat() if seat.created_at else None
            ),
        ))
    return out


async def _software_permissions(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.domains.asset.models import AssetSoftwareCatalog
    from app.domains.permission.models import (
        PERM_ACTIVE,
        SoftwarePermission,
        UserPermission,
    )

    rows = (await db.execute(
        select(UserPermission, SoftwarePermission, AssetSoftwareCatalog)
        .join(
            SoftwarePermission,
            SoftwarePermission.id == UserPermission.permission_id,
        )
        .join(
            AssetSoftwareCatalog,
            AssetSoftwareCatalog.id == SoftwarePermission.software_id,
        )
        .where(
            UserPermission.user_id == person_id,
            UserPermission.status == PERM_ACTIVE,
        )
    )).all()
    return [
        _holding(
            RES_SOFTWARE_PERMISSION, up.id,
            f"{sw.name} — {perm.display_name}",
            software=sw.name,
            permission=perm.display_name,
            policy_target=f"permission:{perm.id}",
            peer_key=f"permission:{perm.id}",
            granted_at=(
                up.granted_at.isoformat() if up.granted_at else None
            ),
        )
        for up, perm, sw in rows
    ]


async def _api_keys(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.domains.apikey.models import ApiKeyAssignment

    rows = (await db.execute(
        select(ApiKeyAssignment).where(
            and_(
                ApiKeyAssignment.assigned_to == person_id,
                ApiKeyAssignment.status.in_(
                    ("active", "approved", "pending_approval"),
                ),
            )
        )
    )).scalars().all()
    return [
        _holding(
            RES_API_KEY, ak.id,
            ak.project_name or ak.reason or str(ak.id),
            status=ak.status,
            granted_at=(
                ak.created_at.isoformat() if ak.created_at else None
            ),
        )
        for ak in rows
    ]


async def _cloud_access_grants(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.domains.cmdb.models import CmdbCloudAccessGrant, CmdbCloudScope

    rows = (await db.execute(
        select(CmdbCloudAccessGrant, CmdbCloudScope)
        .outerjoin(
            CmdbCloudScope,
            CmdbCloudScope.id == CmdbCloudAccessGrant.scope_id,
        )
        .where(
            CmdbCloudAccessGrant.person_id == str(person_id),
            CmdbCloudAccessGrant.status.in_(("active", "pending")),
        )
    )).all()
    return [
        _holding(
            RES_CLOUD_ACCESS_GRANT, grant.id,
            grant.role_display_name or grant.role_key,
            scope=scope.external_id if scope else None,
            role_key=grant.role_key,
            review_status=grant.review_status,
            # No ``policy_target``: provision_policies cannot express a
            # cloud grant, so policies are silent here by construction.
            peer_key=f"cloud:{grant.scope_id}:{grant.role_key}",
            granted_at=(
                grant.granted_at.isoformat() if grant.granted_at else None
            ),
        )
        for grant, scope in rows
    ]


async def _cloud_scope_owner(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.domains.cmdb.models import (
        CmdbCloudScope,
        owner_governed_scope_criteria,
    )

    rows = (await db.execute(
        select(CmdbCloudScope).where(
            CmdbCloudScope.owner_person_id == str(person_id),
            # Shared predicate — a mover cannot hand over a
            # decommissioned account or a cost-allocation row. See
            # ``owner_governed_scope_criteria`` for why each clause is
            # there and why this must not be re-hand-rolled.
            *owner_governed_scope_criteria(),
        )
    )).scalars().all()
    return [
        _holding(
            RES_CLOUD_SCOPE_OWNER, scope.id,
            scope.display_name or scope.external_id,
            scope_type=scope.scope_type,
            external_id=scope.external_id,
        )
        for scope in rows
    ]


async def _cloud_iam_users(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.domains.cmdb.models import CmdbIamUser

    rows = (await db.execute(
        select(CmdbIamUser).where(
            CmdbIamUser.owner_person_id == str(person_id),
            CmdbIamUser.deleted_at.is_(None),
            # Mirrors the offboarding scanner: an already-revoked IAM
            # user's owner FK is bookkeeping, and asking a new manager
            # about it is noise.
            CmdbIamUser.revoked_at.is_(None),
        )
    )).scalars().all()
    return [
        _holding(
            RES_CLOUD_IAM_USER, user.id,
            user.user_name or str(user.id),
            arn=user.arn,
        )
        for user in rows
    ]


def _owner_scan(resource_type: str, model_path: str, label_attr: str):
    """Loader for the "owner_person_id + soft delete" shape, which
    covers five CMDB network-asset tables identically."""

    async def _load(db: AsyncSession, person_id: UUID) -> list[Holding]:
        from app.domains.cmdb import models as cmdb_models

        model = getattr(cmdb_models, model_path)
        rows = (await db.execute(
            select(model).where(
                and_(
                    model.owner_person_id == str(person_id),
                    model.deleted_at.is_(None),
                ),
            )
        )).scalars().all()
        return [
            _holding(
                resource_type, row.id,
                getattr(row, label_attr, None) or str(row.id),
            )
            for row in rows
        ]

    return _load


async def _software_owner(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.domains.asset.models import AssetSoftwareCatalog

    rows = (await db.execute(
        select(AssetSoftwareCatalog).where(
            AssetSoftwareCatalog.owner_person_id == person_id,
        )
    )).scalars().all()
    return [
        _holding(RES_SOFTWARE_OWNER, sw.id, sw.name, status=sw.status)
        for sw in rows
    ]


async def _software_it_contact(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.domains.asset.models import AssetSoftwareCatalog

    rows = (await db.execute(
        select(AssetSoftwareCatalog).where(
            AssetSoftwareCatalog.it_contact_person_id == person_id,
        )
    )).scalars().all()
    return [
        _holding(RES_SOFTWARE_IT_CONTACT, sw.id, sw.name, status=sw.status)
        for sw in rows
    ]


async def _sso_grants(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.domains.sso.models import SsoGrant

    rows = (await db.execute(
        select(SsoGrant).where(
            SsoGrant.person_id == str(person_id),
            SsoGrant.status == "approved",
        )
    )).scalars().all()
    return [
        _holding(
            RES_SSO_GRANT, UUID(str(grant.id)),
            grant.system_name or str(grant.id),
        )
        for grant in rows
    ]


async def _github_org_memberships(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    """Active org memberships this person holds.

    Only ``active`` — a pending invitation is not access yet, and a declined or
    expired one is not access at all. Labelled by ``org/login`` because a
    reviewer needs to know WHICH org and under which GitHub account, and the
    two are not the same fact.
    """
    from app.domains.github_org import constants as gh_const
    from app.domains.github_org.models import GithubOrg, GithubOrgMembership

    rows = (await db.execute(
        select(GithubOrgMembership, GithubOrg)
        .join(GithubOrg, GithubOrg.id == GithubOrgMembership.org_id)
        .where(
            GithubOrgMembership.person_id == str(person_id),
            GithubOrgMembership.state == gh_const.MEMBERSHIP_STATE_ACTIVE,
        )
    )).all()
    return [
        _holding(
            RES_GITHUB_ORG_MEMBERSHIP,
            UUID(str(membership.id)),
            f"{org.slug}/{membership.github_login}",
        )
        for membership, org in rows
    ]


async def _domain_registrations(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.domains.domain_registration.models import DomainRegistration

    rows = (await db.execute(
        select(DomainRegistration).where(
            DomainRegistration.person_id == str(person_id),
            DomainRegistration.status == "approved",
        )
    )).scalars().all()
    return [
        _holding(
            RES_DOMAIN_REGISTRATION, UUID(str(reg.id)),
            reg.domain_name or str(reg.id),
        )
        for reg in rows
    ]


async def _system_owner(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.core.coordinators import system_registry_coordinator as registry

    return [
        _holding(
            RES_SYSTEM_OWNER, UUID(owned["id"]), owned["name"],
            role="primary" if owned["is_primary_owner"] else "backup",
        )
        for owned in await registry.systems_owned_by(db, person_id)
    ]


async def _approval_flow_owner(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.core.coordinators import approval_registry_coordinator as registry

    return [
        _holding(RES_APPROVAL_FLOW_OWNER, UUID(flow["id"]), flow["name"])
        for flow in await registry.flows_owned_by(db, person_id)
    ]


async def _google_orphans(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.domains.google_workspace.models import GoogleOrphanAccount

    rows = (await db.execute(
        select(GoogleOrphanAccount).where(
            GoogleOrphanAccount.owner_user_id == person_id,
            GoogleOrphanAccount.deleted_at.is_(None),
        )
    )).scalars().all()
    return [
        _holding(
            RES_GOOGLE_ORPHAN_OWNER, orphan.id,
            orphan.primary_email or str(orphan.id),
        )
        for orphan in rows
    ]


async def _identity_consumers(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.identity.consumers.models import IdentityConsumer

    rows = (await db.execute(
        select(IdentityConsumer).where(
            IdentityConsumer.owner_person_id == person_id,
            IdentityConsumer.deleted_at.is_(None),
        )
    )).scalars().all()
    return [
        _holding(
            RES_IDENTITY_CONSUMER_OWNER, consumer.id,
            consumer.name or str(consumer.id),
        )
        for consumer in rows
    ]


async def _vpn_applications(db: AsyncSession, person_id: UUID) -> list[Holding]:
    from app.domains.vpn_application.models import (
        VPN_APP_STATUS_APPROVED,
        VpnApplication,
    )

    rows = (await db.execute(
        select(VpnApplication).where(
            VpnApplication.person_id == str(person_id),
            VpnApplication.status == VPN_APP_STATUS_APPROVED,
        )
    )).scalars().all()
    return [
        _holding(
            RES_VPN_APPLICATION, UUID(str(app.id)),
            app.usage_description or app.vpn_node or str(app.id),
        )
        for app in rows
    ]


async def _ai_tool_applications(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    from app.domains.ai_tool_application.models import (
        AI_TOOL_STATUS_APPROVED,
        AiToolApplication,
    )

    rows = (await db.execute(
        select(AiToolApplication).where(
            AiToolApplication.person_id == str(person_id),
            AiToolApplication.status == AI_TOOL_STATUS_APPROVED,
        )
    )).scalars().all()
    return [
        _holding(
            RES_AI_TOOL_APPLICATION, UUID(str(app.id)),
            app.ai_name or str(app.id),
        )
        for app in rows
    ]


async def _data_access_grants(db: AsyncSession, person_id: UUID) -> list[Holding]:
    """Personal BigQuery dataset grants — the same predicate the offboarding
    scanner and the T+1 safety net use (``select_personal_active_bindings``)."""
    from app.domains.data_access.lifecycle import select_personal_active_bindings
    from app.domains.data_access.models import DataAccessResource

    bindings = await select_personal_active_bindings(db, person_id)
    if not bindings:
        return []
    resources = {
        r.id: r for r in (await db.execute(
            select(DataAccessResource).where(
                DataAccessResource.id.in_([b.resource_id for b in bindings]),
            )
        )).scalars().all()
    }
    out: list[Holding] = []
    for b in bindings:
        res = resources.get(b.resource_id)
        locator = ".".join(filter(None, (
            res.project_id if res else None,
            res.dataset_id if res else None,
            res.table_id if res else None,
        ))) or str(b.resource_id)
        out.append(_holding(
            RES_DATA_ACCESS_GRANT, b.id, locator,
            entitlement=b.entitlement_snapshot,
            expires_at=b.expires_at.isoformat() if b.expires_at else None,
            renewal_count=int(b.renewal_count or 0),
        ))
    return out


async def _certificates(db: AsyncSession, person_id: UUID) -> list[Holding]:
    """TLS certificates whose owner was MANUALLY assigned to this person.

    ``owner_is_manual`` is the whole filter: a probed / cloud-synced
    certificate resolves its owner live from the domain chain, so it is
    not this person's to hand over and asking about it would be noise.
    """
    from app.domains.cmdb.models import CmdbCertificate

    rows = (await db.execute(
        select(CmdbCertificate).where(
            and_(
                CmdbCertificate.owner_person_id == str(person_id),
                CmdbCertificate.owner_is_manual.is_(True),
                CmdbCertificate.deleted_at.is_(None),
            ),
        )
    )).scalars().all()
    return [
        _holding(
            RES_CERTIFICATE, cert.id,
            cert.common_name or str(cert.id),
            environment=cert.environment,
        )
        for cert in rows
    ]


async def _email_group_memberships(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    """Static Google-group memberships (ownership is ``_email_groups``).

    Reads the vendor — Google is the source of truth for membership and
    Helix keeps no member mirror. A vendor ERROR is re-raised so the
    packet records the gap in ``failed_types`` instead of quietly showing
    a short list; "no Google integration" / "no work email" are
    determinate empties, not gaps.
    """
    from app.domains.email_group.membership_scan import (
        static_group_memberships_for_person,
    )
    from app.identity.person_models import IdentityPerson

    person = await db.get(IdentityPerson, person_id)
    if person is None:
        return []
    scan = await static_group_memberships_for_person(db, person)
    if scan.skipped_reason and scan.skipped_reason.startswith("google_error"):
        raise RuntimeError(
            f"mail-group membership unreadable: {scan.skipped_reason}",
        )
    owned = {g.id for g in await _owned_group_ids(db, person_id)}
    return [
        _holding(RES_EMAIL_GROUP_MEMBER, g.id, g.email or g.name or str(g.id))
        for g in scan.groups
        if g.id not in owned
    ]


async def _owned_group_ids(db: AsyncSession, person_id: UUID):
    from app.domains.email_group.service import EmailGroupService

    return await EmailGroupService(db).list_groups_owned_by(person_id)


async def _email_groups(db: AsyncSession, person_id: UUID) -> list[Holding]:
    """Google mail groups the person is a registered OWNER of.

    Same call the offboarding scanner makes
    (``EmailGroupService.list_groups_owned_by`` — active groups joined
    through ``email_group_owners``), reached directly rather than via
    ``OffboardingService._scan_email_groups``: that wrapper exists to
    keep the offboarding domain's imports lazy, and routing through a
    12k-line service to borrow one query would drag its constructor into
    a read that needs none of it.

    Google requires at least one owner per group, which is why this is a
    review question and not merely informational: a move that leaves the
    group owned by someone who no longer works on it is how a group ends
    up with an owner nobody can name.
    """
    from app.domains.email_group.service import EmailGroupService

    rows = await EmailGroupService(db).list_groups_owned_by(person_id)
    return [
        _holding(
            RES_EMAIL_GROUP, group.id,
            group.email or group.name or str(group.id),
            name=group.name,
        )
        for group in rows
    ]


async def _software_accounts(
    db: AsyncSession, person_id: UUID,
) -> list[Holding]:
    """Provider account-wide access (currently Jira/Atlassian).

    Deliberately NOT surfaced on a move. The offboarding blocker exists
    because a leaver's access can live in arbitrarily many untracked
    groups and the only reliable action is removing them from every
    group at once — an all-or-nothing action with no "keep some" middle,
    which is precisely the decision a mover review must not force. The
    tracked seats and permissions above already carry the per-product
    questions a move actually raises.

    Kept as an explicit empty loader rather than an omission so the
    coverage guard sees a decision, not a gap.
    """
    return []


_LOADERS = {
    RES_HARDWARE: _hardware,
    RES_SOFTWARE_SEAT: _software_seats,
    RES_GITHUB_ORG_MEMBERSHIP: _github_org_memberships,
    RES_SOFTWARE_PERMISSION: _software_permissions,
    RES_SOFTWARE_ACCOUNT: _software_accounts,
    RES_API_KEY: _api_keys,
    RES_CLOUD_ACCESS_GRANT: _cloud_access_grants,
    RES_CLOUD_SCOPE_OWNER: _cloud_scope_owner,
    RES_CLOUD_IAM_USER: _cloud_iam_users,
    RES_DOMAIN: _owner_scan(RES_DOMAIN, "CmdbDomain", "domain_name"),
    RES_DNS_RECORD: _owner_scan(RES_DNS_RECORD, "CmdbDnsRecord", "name"),
    RES_CLOUD_SERVER: _owner_scan(RES_CLOUD_SERVER, "CmdbServer", "name"),
    RES_LOAD_BALANCER: _owner_scan(
        RES_LOAD_BALANCER, "CmdbLoadBalancer", "name",
    ),
    RES_CDN_DOMAIN: _owner_scan(RES_CDN_DOMAIN, "CmdbCdnDomain", "domain_name"),
    RES_SOFTWARE_OWNER: _software_owner,
    RES_SOFTWARE_IT_CONTACT: _software_it_contact,
    RES_SSO_GRANT: _sso_grants,
    RES_DOMAIN_REGISTRATION: _domain_registrations,
    RES_SYSTEM_OWNER: _system_owner,
    RES_APPROVAL_FLOW_OWNER: _approval_flow_owner,
    RES_GOOGLE_ORPHAN_OWNER: _google_orphans,
    RES_IDENTITY_CONSUMER_OWNER: _identity_consumers,
    RES_VPN_APPLICATION: _vpn_applications,
    RES_AI_TOOL_APPLICATION: _ai_tool_applications,
    RES_EMAIL_GROUP: _email_groups,
    RES_DATA_ACCESS_GRANT: _data_access_grants,
    RES_OBJECT_STORAGE: _owner_scan(
        RES_OBJECT_STORAGE, "CmdbObjectStorage", "bucket_name",
    ),
    RES_CERTIFICATE: _certificates,
    RES_EMAIL_GROUP_MEMBER: _email_group_memberships,
}


__all__ = [
    "INFORMATIONAL",
    "RESOURCE_DISPOSITIONS",
    "REVIEW",
    "Holding",
    "HoldingsScan",
    "scan_person_holdings",
]