"""Static Google-group memberships a person holds (访问对象闭环 PR3).

The offboarding flow has always handled a leaver's group OWNERSHIP
(``RES_EMAIL_GROUP`` → transfer). Plain membership was invisible: the
leaver stayed on ``finance-team@`` as a member, which is harmless while
their Google account is suspended and wrong the moment it is retained
(CBG / shared-service classification) or revived.

Two entry points, both read the vendor (Google is the source of truth for
membership — Helix keeps no member mirror, only ``direct_members_count``):

* :func:`static_group_memberships_for_person` — the groups (as local
  ``EmailGroup`` rows) the person is a member of. Dynamic groups are
  excluded (membership is rule-computed; the person drops out when the
  rule stops matching). Groups Google returns that Helix has not mirrored
  are reported in ``unmirrored`` so the scan summary can say so.
* :func:`remove_person_from_group` — the T+0 / manual revoke.

A missing Google integration is a *skip*, not an error: the scanner
records ``skipped_reason`` and the blocker family simply does not appear.
"""
from __future__ import annotations

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

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

from app.domains.email_group.models import EmailGroup, EmailGroupOwner
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)


@dataclass(slots=True)
class MembershipScan:
    groups: list[EmailGroup] = field(default_factory=list)
    #: Google groups the person is in that Helix has no local row for.
    unmirrored: list[str] = field(default_factory=list)
    #: ``None`` when the roster was read; a reason string when it was not.
    skipped_reason: str | None = None


async def _google_driver(db: AsyncSession) -> Any | None:
    from app.integrations.drivers.google_workspace import (  # noqa: PLC0415
        GoogleWorkspaceDriver,
    )
    from app.integrations.models import Integration  # noqa: PLC0415
    from app.integrations.service import _resolve  # noqa: PLC0415

    row = (await db.execute(
        select(Integration).where(
            Integration.kind == "software_provisioning",
            Integration.provider == "google_workspace",
            Integration.enabled.is_(True),
        ).order_by(Integration.is_primary.desc(), Integration.created_at)
    )).scalars().first()
    if row is None:
        return None
    resolved = await _resolve(db, row)
    return GoogleWorkspaceDriver(integration_id=str(row.id), config=resolved.config)


async def static_group_memberships_for_person(
    db: AsyncSession,
    person: IdentityPerson,
    *,
    driver: Any | None = None,
) -> MembershipScan:
    """Groups ``person`` is a member of, as local rows (dynamic groups excluded)."""
    scan = MembershipScan()
    email = (person.work_email or "").strip().lower()
    if not email:
        scan.skipped_reason = "no_work_email"
        return scan
    drv = driver if driver is not None else await _google_driver(db)
    if drv is None:
        scan.skipped_reason = "google_not_configured"
        return scan
    try:
        # ``raise_on_missing=True`` on purpose. The default swallows Google's
        # 404 (user not in the directory) into ``[]`` — right for the "我的邮件组"
        # tab, fatally wrong here: a leaver whose Workspace account is already
        # suspended or deleted is exactly the 404 case, and an empty list would
        # be read as "member of nothing" and close the family clean. A skip
        # says "not read"; ``[]`` says "read, and there was nothing".
        raw_groups = await drv.list_groups_for_user(email, raise_on_missing=True)
    except Exception as exc:  # noqa: BLE001 — the caller decides how loud to be
        logger.warning("membership_scan: groups.list(userKey=%s) failed: %s", email, exc)
        scan.skipped_reason = f"google_error:{type(exc).__name__}"
        return scan
    emails = sorted({
        str(g.get("email") or "").strip().lower()
        for g in raw_groups if isinstance(g, dict) and g.get("email")
    })
    if not emails:
        return scan
    local = {
        (g.email or "").lower(): g
        for g in (await db.execute(
            select(EmailGroup).where(func.lower(EmailGroup.email).in_(emails)),
        )).scalars().all()
    }
    for e in emails:
        g = local.get(e)
        if g is None:
            scan.unmirrored.append(e)
            continue
        if g.is_dynamic:
            continue
        if (g.status or "active") != "active":
            continue
        scan.groups.append(g)
    return scan


async def person_owns_group(db: AsyncSession, *, group_id: uuid.UUID, person_id: uuid.UUID) -> bool:
    hit = await db.scalar(
        select(EmailGroupOwner.id).where(
            EmailGroupOwner.email_group_id == group_id,
            EmailGroupOwner.person_id == person_id,
        ).limit(1),
    )
    return hit is not None


async def remove_person_from_group(
    db: AsyncSession,
    *,
    group: EmailGroup,
    person: IdentityPerson,
    trace_id: str,
    actor_user_id: uuid.UUID | None,
    reason: str,
    driver: Any | None = None,
) -> dict[str, Any]:
    """``members.delete`` for a leaver / mover. Idempotent on an already-absent member.

    Refuses while the person is still an OWNER of the group in Helix's
    mirror: removing the last owner breaks Google's ≥1-owner invariant and
    the ownership transfer (``RES_EMAIL_GROUP``) must land first.
    """
    from app.audit.service import AuditService  # noqa: PLC0415
    from app.core.exceptions import AppException  # noqa: PLC0415

    if await person_owns_group(db, group_id=group.id, person_id=person.id):
        raise AppException(
            code="EMAIL_GROUP_MEMBER_STILL_OWNER",
            message=(
                f"{person.work_email} still owns {group.email}; transfer the "
                "ownership first (RES_EMAIL_GROUP), then remove the membership."
            ),
            status_code=409,
        )
    drv = driver if driver is not None else await _google_driver(db)
    if drv is None:
        raise AppException(
            code="GOOGLE_WORKSPACE_NOT_CONFIGURED",
            message="Google Workspace integration is not configured.",
            status_code=503,
        )
    result = await drv.remove_group_member(
        group_email=group.email, member_email=(person.work_email or "").strip().lower(),
    )
    if not getattr(result, "ok", False):
        # Propagate: the T+0 savepoint records the error on the blocker and
        # retries next tick; a silent "removed" here would be the 2026-08-13
        # failure shape all over again (Helix says gone, vendor says no).
        raise AppException(
            code="EMAIL_GROUP_MEMBER_REMOVE_FAILED",
            message=(
                f"Google refused to remove {person.work_email} from {group.email}: "
                f"{getattr(result, 'error_message', None) or 'unknown error'}"
            ),
            status_code=502,
        )
    outcome = {
        "group": group.email,
        "member": person.work_email,
        "status": "removed",
        "data": dict(getattr(result, "data", None) or {}),
    }
    await AuditService(db).log(
        trace_id=trace_id,
        actor_type="user" if actor_user_id else "system",
        actor_user_id=actor_user_id,
        action="email_group.member_removed",
        status="success",
        resource_type="email_group",
        resource_id=str(group.id),
        detail={**outcome, "person_id": str(person.id), "reason": reason},
    )
    return outcome


__all__ = [
    "MembershipScan",
    "person_owns_group",
    "remove_person_from_group",
    "static_group_memberships_for_person",
]