"""EmailGroupManageService — the admin-tier view over the whole mail-group
estate (HR / Global Admin).

This is the read side of the 邮件组管理 desk. It exists next to (not
inside) :class:`~app.domains.email_group.service.EmailGroupService`
because the two answer different questions:

* ``EmailGroupService`` is the *actor-scoped* surface — "which groups may
  **I** see / delete / manage members of", with owner-OR-admin checks per
  operation. It backs the public directory and the self-service flows.
* This service is the *estate-scoped* surface — "show me all 760 groups
  with governance signals, and show me everything that has ever happened
  to them". Every method is a flat ``can_manage_email_groups`` role gate;
  there is no per-group ownership path in here at all.

Both flavours of group (static + rule-driven dynamic) are listed through
one query so the management page can show the estate as a whole and
filter by kind, rather than making HR guess which of two menus a group
lives under.
"""
from __future__ import annotations

import logging
from collections.abc import Sequence
from datetime import datetime
from typing import Any
from uuid import UUID

from sqlalchemy import String as SaString
from sqlalchemy import and_, asc, case, desc, func, or_, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.audit.models import AuditLog
from app.audit.service import _esc
from app.core.exceptions import AppException, NotFoundError
from app.core.rbac import can_manage_email_groups
from app.domains.email_group.manage_actions import category_of, known_actions
from app.domains.email_group.models import EmailGroup, EmailGroupOwner
from app.identity.person_models import IdentityPerson

logger = logging.getLogger(__name__)

# Google mirrors every group address into a `<addr>@<domain>.test-google-a.com`
# shadow alias. It is not an address anyone can mail and it exists on
# essentially every group, so surfacing it as "this group has an alias"
# would make the alias column pure noise (131 of 133 cleanup candidates
# carry one and nothing else). Filtered out of every alias list this
# service returns; `aliases_raw` keeps the unfiltered mirror for anyone
# who needs the literal Google state.
_GOOGLE_TEST_ALIAS_SUFFIX = ".test-google-a.com"

# Sort keys the list endpoint accepts → the column to order by. A
# whitelist (not a getattr on user input) so no request can order by an
# arbitrary column.
_SORT_COLUMNS = {
    "email": EmailGroup.email,
    "name": EmailGroup.name,
    "members": EmailGroup.direct_members_count,
    "created_at": EmailGroup.created_at,
    "last_synced_at": EmailGroup.last_synced_at,
}

# Free-text group search resolves to a set of group ids that then filter
# ``audit_logs.resource_id``. Cap the id set so a one-character search
# can't build a 700-element IN-list; past the cap we tell the caller to
# narrow instead of silently truncating (a truncated IN-list would show
# a *partial* log and look like "nothing happened to that group").
_AUDIT_GROUP_SEARCH_CAP = 200

# Largest page an interactive request may ask for. The export paths pass
# their own higher ``max_page_size``; see ``list_groups``.
_API_PAGE_MAX = 200

# Only actions in this domain's namespace belong in this domain's log —
# see the ``audit_log`` filter for why resource_type alone is not enough.
_ACTION_PREFIX = "email_group."


def _like(term: str) -> str:
    """``%term%`` with LIKE metacharacters escaped.

    Values are always bound as parameters (no injection risk either
    way), but an unescaped ``_`` or ``%`` typed into a search box would
    silently widen the match — ``g_all`` would hit ``g-all`` — which
    reads as a bug in the filter. Escaping comes from
    :func:`app.audit.service._esc` rather than a second copy of the
    same three ``replace`` calls. Every call site pairs it with
    ``escape="\\"``, per the convention the other LIKE helpers in this
    repo follow.
    """
    return f"%{_esc(term.strip().lower())}%"


def real_aliases(aliases: list[str] | None) -> list[str]:
    """Aliases with Google's test-domain mirrors removed, sorted.

    ``None`` (never synced) and ``[]`` (synced, none) both collapse to
    ``[]`` here — callers that need to tell those apart read the row's
    ``aliases`` column directly.
    """
    return sorted(
        a for a in (aliases or [])
        if a and not a.endswith(_GOOGLE_TEST_ALIAS_SUFFIX)
    )


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

    # ── gate ──────────────────────────────────────────────────────

    def require_manage(self, actor_roles: list[str]) -> None:
        if not can_manage_email_groups(actor_roles):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有 HR / IT Admin 才能管理邮件组。",
                status_code=403,
            )

    # ── overview ──────────────────────────────────────────────────

    async def overview(self, *, actor_roles: list[str]) -> dict[str, Any]:
        """Estate-level counters for the management page's KPI strip.

        Every number is computed in ONE pass over ``email_groups`` with
        conditional aggregates plus one correlated owner-count subquery,
        rather than N COUNT queries — the page loads all of them at once
        on every tab switch.

        ``unowned_with_members`` is the governance number that matters:
        a group with members but nobody accountable for it. Groups with
        *zero* members and no owner are the cleanup pool instead, and
        the two sets are disjoint by construction (Google counts owners
        as direct members, so ``direct_members_count = 0`` already
        implies "no owner").
        """
        self.require_manage(actor_roles)

        owned = select(EmailGroupOwner.email_group_id).distinct().scalar_subquery()
        is_active = EmailGroup.status == "active"
        is_dynamic = EmailGroup.is_dynamic.is_(True)
        # NOT dynamic = explicitly false. ``NULL`` means "Cloud Identity
        # detection has never run determinately for this row" and must
        # NOT be counted as static — treating an undetermined row as
        # static is how a rule-driven group would end up in a static
        # cleanup pool.
        is_static = EmailGroup.is_dynamic.is_(False)
        has_owner = EmailGroup.id.in_(owned)
        zero_members = EmailGroup.direct_members_count == 0
        # Spelled as ``> 0`` rather than ``NOT (= 0)``: an unsynced row
        # (``direct_members_count IS NULL``) has UNKNOWN membership, and
        # ``NOT (NULL = 0)`` is NULL, so a negation would quietly drop
        # it from both buckets. ``> 0`` states the intent directly —
        # only rows we positively know have members.
        has_members = EmailGroup.direct_members_count > 0

        def _count_if(*conditions: Any):
            """``SUM(CASE WHEN … THEN 1 ELSE 0 END)`` as one aggregate.

            ``CASE`` (not MySQL's ``IF()``) so the same statement runs
            on the SQLite used by some unit tests as well as on
            MySQL/TiDB.
            """
            return func.sum(case((and_(*conditions), 1), else_=0))

        row = (await self.db.execute(
            select(
                func.count(EmailGroup.id).label("total"),
                _count_if(is_active).label("active"),
                _count_if(is_active, is_static).label("static"),
                _count_if(is_active, is_dynamic).label("dynamic"),
                _count_if(
                    is_active, EmailGroup.is_dynamic.is_(None),
                ).label("kind_unknown"),
                _count_if(
                    is_active, ~has_owner, has_members,
                ).label("unowned_with_members"),
                _count_if(is_active, ~has_owner).label("unowned"),
                _count_if(
                    is_active, is_static, zero_members,
                ).label("static_zero_members"),
                _count_if(EmailGroup.status == "deleted").label("deleted"),
                func.max(EmailGroup.last_synced_at).label("last_synced_at"),
            )
        )).mappings().one()

        def _n(key: str) -> int:
            # ``SUM`` returns Decimal on MySQL/TiDB and NULL on an empty
            # table — normalise both to a plain int.
            return int(row[key] or 0)

        return {
            "total": _n("total"),
            "active": _n("active"),
            "static": _n("static"),
            "dynamic": _n("dynamic"),
            "kind_unknown": _n("kind_unknown"),
            "unowned_with_members": _n("unowned_with_members"),
            "unowned": _n("unowned"),
            "static_zero_members": _n("static_zero_members"),
            "deleted": _n("deleted"),
            "last_synced_at": row["last_synced_at"],
        }

    # ── group list ────────────────────────────────────────────────

    def _list_conditions(
        self,
        *,
        search: str | None,
        suffix: str | None,
        kind: str,
        members: str,
        owners: str,
        include_deleted: bool,
    ) -> list[Any]:
        conds: list[Any] = []
        if include_deleted:
            # Deleted rows are tombstones kept so audit / offboarding
            # references stay resolvable. Showing them is opt-in.
            conds.append(EmailGroup.status.in_(("active", "deleted")))
        else:
            conds.append(EmailGroup.status == "active")

        if search:
            pattern = _like(search)
            # Aliases are a JSON array column; CAST-to-CHAR + LIKE is the
            # portable form that behaves the same on MySQL 8 and TiDB
            # (JSON_SEARCH's argument shapes differ subtly between them).
            aliases_text = func.lower(
                func.coalesce(func.cast(EmailGroup.aliases, SaString), "[]"),
            )
            conds.append(or_(
                func.lower(EmailGroup.email).like(pattern, escape="\\"),
                func.lower(EmailGroup.name).like(pattern, escape="\\"),
                func.lower(func.coalesce(EmailGroup.description, "")).like(
                    pattern, escape="\\",
                ),
                aliases_text.like(pattern, escape="\\"),
            ))

        if suffix:
            clean = suffix.strip().lower()
            if clean:
                # Anchor on the right so ``pingcap.com`` does not match
                # ``pingcap.com.cn``. Escaped like every other LIKE here:
                # ``suffix`` is a free query param, and an unescaped
                # ``_`` in it is a single-character wildcard, so
                # ``ping_ap.com`` would quietly match ``pingcap.com``
                # and the filter would look broken rather than empty.
                conds.append(
                    func.lower(EmailGroup.email).like(
                        f"%@{_esc(clean)}", escape="\\",
                    )
                )

        if kind == "static":
            conds.append(EmailGroup.is_dynamic.is_(False))
        elif kind == "dynamic":
            conds.append(EmailGroup.is_dynamic.is_(True))
        elif kind == "unknown":
            conds.append(EmailGroup.is_dynamic.is_(None))

        if members == "zero":
            conds.append(EmailGroup.direct_members_count == 0)
        elif members == "nonzero":
            conds.append(EmailGroup.direct_members_count > 0)
        elif members == "unsynced":
            conds.append(EmailGroup.direct_members_count.is_(None))

        if owners in ("none", "some"):
            owned = select(EmailGroupOwner.email_group_id).distinct().scalar_subquery()
            if owners == "none":
                conds.append(~EmailGroup.id.in_(owned))
            else:
                conds.append(EmailGroup.id.in_(owned))

        return conds

    async def list_groups(
        self,
        *,
        actor_roles: list[str],
        search: str | None = None,
        suffix: str | None = None,
        kind: str = "all",
        members: str = "all",
        owners: str = "all",
        include_deleted: bool = False,
        sort: str = "email",
        order: str = "asc",
        page: int = 1,
        page_size: int = 50,
        max_page_size: int = _API_PAGE_MAX,
    ) -> tuple[list[dict[str, Any]], int]:
        """Paginated management list. Returns ``(rows, total)``.

        ``max_page_size`` exists so the export path can ask for more
        than an interactive page. Clamping unconditionally to
        ``_API_PAGE_MAX`` made the export's own cap constant dead and
        shipped 200 of 760 groups while reporting 760 as the total —
        loudly (the truncation header fires) but still wrong.
        """
        self.require_manage(actor_roles)
        page = max(page, 1)
        page_size = max(min(page_size, max_page_size), 1)

        conds = self._list_conditions(
            search=search, suffix=suffix, kind=kind, members=members,
            owners=owners, include_deleted=include_deleted,
        )

        sort_col = _SORT_COLUMNS.get(sort, EmailGroup.email)
        direction = desc if order == "desc" else asc
        # Always tie-break on email: ``direct_members_count`` and
        # ``last_synced_at`` have thousands of ties, and an unstable sort
        # makes rows jump between pages as the user paginates.
        order_by = [direction(sort_col)]
        if sort != "email":
            order_by.append(asc(EmailGroup.email))

        total = int((await self.db.execute(
            select(func.count(EmailGroup.id)).where(*conds)
        )).scalar() or 0)

        rows = (await self.db.execute(
            select(EmailGroup)
            .where(*conds)
            .options(selectinload(EmailGroup.owners))
            .order_by(*order_by)
            .offset((page - 1) * page_size)
            .limit(page_size)
        )).scalars().all()

        persons = await self._resolve_owner_persons(rows)
        return [self.serialize_group(g, persons) for g in rows], total

    async def _resolve_owner_persons(
        self, groups: Sequence[EmailGroup],
    ) -> dict[UUID, IdentityPerson]:
        person_ids = {o.person_id for g in groups for o in g.owners}
        if not person_ids:
            return {}
        found = (await self.db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(person_ids))
        )).scalars().all()
        return {p.id: p for p in found}

    @staticmethod
    def serialize_group(
        group: EmailGroup, persons: dict[UUID, IdentityPerson],
    ) -> dict[str, Any]:
        owners = []
        for o in group.owners:
            p = persons.get(o.person_id)
            owners.append({
                "person_id": str(o.person_id),
                "display_name": p.display_name if p else None,
                "work_email": p.work_email if p else None,
                "employee_no": p.employee_no if p else None,
                "employment_status": p.employment_status if p else None,
            })
        owners.sort(key=lambda o: (o["display_name"] or o["work_email"] or ""))

        # ``kind`` collapses the tri-state ``is_dynamic`` into the word
        # the UI filters and labels by. ``unknown`` is a real state, not
        # a default: it means Cloud Identity detection has never run
        # determinately (SA scope missing), and it must stay visible so
        # nobody treats such a group as confirmed-static.
        if group.is_dynamic is True:
            kind = "dynamic"
        elif group.is_dynamic is False:
            kind = "static"
        else:
            kind = "unknown"

        return {
            "id": str(group.id),
            "email": group.email,
            "name": group.name,
            "description": group.description,
            "status": group.status,
            "kind": kind,
            "google_group_id": group.google_group_id,
            "direct_members_count": group.direct_members_count,
            "aliases": real_aliases(group.aliases),
            "aliases_raw": sorted(group.aliases or []),
            "alias_never_synced": group.aliases is None,
            "owners": owners,
            "owner_count": len(owners),
            "membership_query": group.membership_query,
            "dynamic_dead_values": group.dynamic_dead_values or [],
            "created_at": group.created_at,
            "updated_at": group.updated_at,
            "last_synced_at": group.last_synced_at,
        }

    # ── operation log ─────────────────────────────────────────────

    async def get_group(self, group_id: UUID, *, actor_roles: list[str]) -> dict[str, Any]:
        """One group by id, serialized — gate included.

        The gate lives HERE, not in the router body, which is the
        convention every other read on this surface follows and the one
        this repo has already paid for breaking: a check in a router
        body is skipped entirely on an ``@idempotent`` replay, and it is
        invisible to any second caller of the same logic. This endpoint
        returns the full owner roster (display name, work email,
        employee number, employment status) for an arbitrary id
        including tombstones, so the check must travel with the query
        rather than with the route.

        Includes soft-deleted rows: an operator opening the drawer on a
        tombstone should see the tombstone, not a 404.
        """
        self.require_manage(actor_roles)
        group = (await self.db.execute(
            select(EmailGroup)
            .where(EmailGroup.id == group_id)
            .options(selectinload(EmailGroup.owners))
        )).scalar_one_or_none()
        if group is None:
            raise NotFoundError("邮件组不存在。")
        persons: dict[UUID, IdentityPerson] = {}
        if group.owners:
            rows = (await self.db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_([o.person_id for o in group.owners])
                )
            )).scalars().all()
            persons = {p.id: p for p in rows}
        return self.serialize_group(group, persons)

    async def audit_log(
        self,
        *,
        actor_roles: list[str],
        group_id: UUID | None = None,
        search: str | None = None,
        actions: list[str] | None = None,
        category: str | None = None,
        actor_person_id: UUID | None = None,
        actor_type: str | None = None,
        status: str | None = None,
        date_from: datetime | None = None,
        date_to: datetime | None = None,
        page: int = 1,
        page_size: int = 20,
        max_page_size: int = _API_PAGE_MAX,
    ) -> tuple[list[dict[str, Any]], int]:
        """The 操作日志 tab: every audited mail-group event, filterable.

        Reads ``audit_logs`` where ``resource_type='email_group'`` — the
        domain has written that row for every mutation since it shipped,
        so this endpoint is a *view*, not a new log. Rows are enriched
        on read (actor identity, group email) so a rename in the
        directory surfaces immediately while the audit rows themselves
        stay immutable.

        ``search`` matches the GROUP (email / name / alias), resolved to
        an id set that filters ``resource_id`` — precise and index-
        friendly, unlike a LIKE over the ``detail`` JSON blob. Deleted
        groups resolve too, which is the whole point of soft-delete:
        the log for a group cleaned up last month is still searchable by
        the address it used to have.
        """
        self.require_manage(actor_roles)
        page = max(page, 1)
        page_size = max(min(page_size, max_page_size), 1)

        # BOTH predicates are required. ``resource_type='email_group'``
        # alone is not this domain's log: ``data_access`` writes it with
        # ``action='data_access.binding.*'`` whose detail carries
        # BigQuery project / dataset ids and per-principal role grants,
        # and ``soc2_email_group`` writes it too. This surface is
        # readable by plain HR and its export serialises ``detail``
        # verbatim, so filtering on resource_type only would hand HR
        # another domain's topology. The action prefix is what makes the
        # log mean "things the mail-group desk did".
        conds: list[Any] = [
            AuditLog.resource_type == "email_group",
            # ``autoescape``: the prefix itself contains an
            # underscore, which is LIKE's single-character wildcard, so
            # the default renders ``LIKE 'email_group.%'`` and would
            # also admit an ``emailXgroup.`` namespace. Harmless today
            # (``resource_type`` pins it) but the kind of thing that
            # becomes a real leak the moment someone reuses this filter
            # without the resource_type condition.
            AuditLog.action.startswith(_ACTION_PREFIX, autoescape=True),
        ]

        if group_id is not None:
            conds.append(AuditLog.resource_id == str(group_id))
        elif search:
            ids = await self._resolve_group_ids_for_search(search)
            if ids is None:
                raise AppException(
                    code="SEARCH_TOO_BROAD",
                    message=(
                        f"该关键词匹配到超过 {_AUDIT_GROUP_SEARCH_CAP} 个邮件组,"
                        "请输入更精确的邮箱或名称。"
                    ),
                    status_code=400,
                )
            if not ids:
                # No group matches → no rows. Return early rather than
                # emitting ``IN ()`` (invalid SQL on MySQL/TiDB).
                return [], 0
            conds.append(AuditLog.resource_id.in_(ids))

        if actions:
            if len(actions) == 1:
                conds.append(AuditLog.action == actions[0])
            else:
                conds.append(AuditLog.action.in_(actions))
        elif category == "other":
            # ``category_of`` returns ``other`` for anything the registry
            # does not list, and the list renders those rows — so the
            # filter has to be able to select them. Resolving ``other``
            # to a registry subset (there is none) would have returned
            # zero rows, i.e. applying ANY category filter silently hid
            # every unregistered action.
            conds.append(AuditLog.action.notin_(known_actions()))
        elif category:
            in_cat = [a for a in known_actions() if category_of(a) == category]
            if not in_cat:
                return [], 0
            conds.append(AuditLog.action.in_(in_cat))

        if actor_person_id is not None:
            conds.append(AuditLog.actor_user_id == actor_person_id)
        if actor_type:
            conds.append(AuditLog.actor_type == actor_type)
        if status:
            conds.append(AuditLog.status == status)
        if date_from is not None:
            conds.append(AuditLog.created_at >= date_from)
        if date_to is not None:
            conds.append(AuditLog.created_at <= date_to)

        total = int((await self.db.execute(
            select(func.count(AuditLog.id)).where(*conds)
        )).scalar() or 0)

        rows = (await self.db.execute(
            select(AuditLog)
            .where(*conds)
            # id tie-break: several rows of one batch share a created_at
            # to the millisecond, and an unstable order duplicates /
            # drops rows across pages.
            .order_by(desc(AuditLog.created_at), desc(AuditLog.id))
            .offset((page - 1) * page_size)
            .limit(page_size)
        )).scalars().all()

        return await self.enrich_audit_rows(list(rows)), total

    async def _resolve_group_ids_for_search(
        self, search: str,
    ) -> list[str] | None:
        """Group ids whose email / name / alias matches ``search``.

        Returns ``None`` when the match set exceeds
        :data:`_AUDIT_GROUP_SEARCH_CAP` so the caller can ask for a
        narrower term instead of showing a silently-truncated log.
        Includes soft-deleted groups.
        """
        pattern = _like(search)
        aliases_text = func.lower(
            func.coalesce(func.cast(EmailGroup.aliases, SaString), "[]"),
        )
        ids = (await self.db.execute(
            select(EmailGroup.id).where(or_(
                func.lower(EmailGroup.email).like(pattern, escape="\\"),
                func.lower(EmailGroup.name).like(pattern, escape="\\"),
                aliases_text.like(pattern, escape="\\"),
            )).limit(_AUDIT_GROUP_SEARCH_CAP + 1)
        )).scalars().all()
        if len(ids) > _AUDIT_GROUP_SEARCH_CAP:
            return None
        return [str(i) for i in ids]

    async def enrich_audit_rows(
        self, rows: list[AuditLog],
    ) -> list[dict[str, Any]]:
        """Attach actor identity + group label to each audit row.

        Two batch lookups regardless of page size (no N+1). The group
        label prefers the LIVE ``email_groups`` row and falls back to
        ``detail.group_email`` — which is what makes a hard-deleted
        group's rows still readable, and what covers rows written before
        a group row existed.
        """
        if not rows:
            return []

        actor_ids = {r.actor_user_id for r in rows if r.actor_user_id}
        actors: dict[str, IdentityPerson] = {}
        if actor_ids:
            found = (await self.db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(actor_ids))
            )).scalars().all()
            actors = {str(p.id): p for p in found}

        # ``resource_id`` is a free-form string column; only rows whose
        # value parses as a UUID can name an email_groups row.
        group_ids: set[UUID] = set()
        for r in rows:
            if not r.resource_id:
                continue
            try:
                group_ids.add(UUID(r.resource_id))
            except (ValueError, AttributeError, TypeError):
                continue
        groups: dict[str, EmailGroup] = {}
        if group_ids:
            found_groups = (await self.db.execute(
                select(EmailGroup).where(EmailGroup.id.in_(group_ids))
            )).scalars().all()
            groups = {str(g.id): g for g in found_groups}

        out: list[dict[str, Any]] = []
        for r in rows:
            actor = actors.get(str(r.actor_user_id)) if r.actor_user_id else None
            group = groups.get(r.resource_id or "")
            detail = r.detail if isinstance(r.detail, dict) else None
            out.append({
                "id": str(r.id),
                "trace_id": str(r.trace_id),
                "created_at": r.created_at,
                "action": r.action,
                "action_category": category_of(r.action),
                "status": r.status,
                "actor_type": r.actor_type,
                "actor_user_id": str(r.actor_user_id) if r.actor_user_id else None,
                "actor_display_name": actor.display_name if actor else None,
                "actor_email": actor.work_email if actor else None,
                "actor_employee_no": actor.employee_no if actor else None,
                "actor_employment_status": (
                    actor.employment_status if actor else None
                ),
                "group_id": r.resource_id,
                "group_email": (
                    group.email if group
                    else (detail or {}).get("group_email")
                ),
                "group_name": group.name if group else None,
                "group_status": group.status if group else None,
                "ip_address": r.ip_address,
                "detail": detail,
            })
        return out