"""Email group REST API.

Layout:

* ``GET  /email-groups``                — public directory (any logged-in user)
* ``GET  /email-groups/suffixes``       — allowed domain suffixes
* ``POST /email-groups/requests``       — self-service create
* ``DELETE /email-groups/{id}``         — owner / HR / IT Admin
* ``POST /email-groups/{id}/transfer-owner`` — owner / HR / IT Admin
* ``POST /email-groups/aliases``        — self-service alias
* ``POST /email-groups/password-reset`` — self-service password reset
* ``POST /email-groups/password-reset/admin`` — IT Admin resets another user
"""
from __future__ import annotations

from datetime import UTC, date, datetime, time
from uuid import UUID

from fastapi import APIRouter, Depends, Query, Request
from pydantic import BaseModel, Field
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    check_sensitive_endpoint_rate_limit,
    get_current_user,
    require_admin,
)
from app.core.dependencies import (  # noqa: E402
    require_active_employee_for_mutations as _leaver_runtime_gate,
)
from app.core.exceptions import NotFoundError
from app.core.idempotent_route import idempotent
from app.core.response import paginated_response, success_response
from app.domains.email_group.manage_service import real_aliases
from app.domains.email_group.models import EmailGroup, EmailGroupOwner
from app.domains.email_group.schemas import (
    EmailAliasCreateRequest,
    EmailGroupCreateRequest,
    EmailGroupOwnerDirectoryEntry,
    EmailGroupOwnerResponse,
    EmailGroupResponse,
    EmailGroupTransferOwnerRequest,
    GroupMemberRemoveRequest,
    JoinDecisionRequest,
    JoinGroupRequest,
    PasswordResetAdminRequest,
    PasswordResetRequest,
    SuffixWhitelistResponse,
)
from app.domains.email_group.service import EmailGroupService
from app.identity.person_models import IdentityPerson

router = APIRouter(
    prefix="/api/v1/email-groups",
    tags=["email_groups"],
    dependencies=[Depends(_leaver_runtime_gate)],
)


def _parse_day_bound(value: date | None, *, end_of_day: bool) -> datetime | None:
    """Convert a ``YYYY-MM-DD`` query param to a timezone-aware datetime.

    Start bound uses 00:00:00; end bound uses 23:59:59.999999 so the
    range is inclusive on both ends in calendar-day terms. The service
    layer stores ``created_at`` as UTC, so we return UTC-aware.
    """
    if value is None:
        return None
    if end_of_day:
        return datetime.combine(value, time.max, tzinfo=UTC)
    return datetime.combine(value, time.min, tzinfo=UTC)


def _serialize_group(group, persons_map):
    owners = []
    for o in group.owners:
        person = persons_map.get(o.person_id)
        owners.append(
            EmailGroupOwnerResponse(
                person_id=o.person_id,
                display_name=person.display_name if person else None,
                work_email=person.work_email if person else None,
                display_name_employee_no=person.employee_no if person else None,
                display_name_employment_status=(
                    person.employment_status if person else None
                ),
            )
        )
    # Aliases land as a JSON array on the wire — coerce ``None``
    # (= "never synced") to ``[]`` so the UI can iterate uniformly. The
    # member-count stays nullable so the UI can distinguish unsynced
    # rows from real "0 members" groups.
    return EmailGroupResponse(
        id=group.id,
        email=group.email,
        name=group.name,
        description=group.description,
        status=group.status,
        owners=owners,
        # ``real_aliases``, not ``sorted(... or [])``: this tenant's
        # Google test-domain mirrors (``<addr>.test-google-a.com``) are
        # merged into the column by the sync, and listing them to every
        # employee as if they were real second addresses is how somebody
        # ends up mailing one. (The NULL-vs-``[]`` distinction is not
        # exposed on this all-employee read — it is an operator concern,
        # surfaced on the 邮件组管理 desk instead.)
        aliases=real_aliases(group.aliases),
        direct_members_count=group.direct_members_count,
        created_at=group.created_at,
        updated_at=group.updated_at,
    )


# ── Reads ──

@router.get("/suffixes")
async def list_suffixes(
    request: Request,
    _current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    suffixes = await service.get_suffix_whitelist()
    return success_response(
        data=SuffixWhitelistResponse(suffixes=suffixes), request=request,
    )


@router.get("")
async def list_groups(
    request: Request,
    search: str | None = Query(None, max_length=255),
    suffix: str | None = Query(
        None, max_length=253,
        description="Exact domain after @ (e.g. pingcap.com)",
    ),
    owner_person_id: UUID | None = Query(
        None,
        description="Only groups where this person is an owner.",
    ),
    only_mine: bool = Query(
        False,
        description=(
            "Shortcut for owner_person_id=<current user>; takes"
            " precedence over owner_person_id if both are set."
        ),
    ),
    only_unowned: bool = Query(
        False,
        description="Governance filter: groups with zero owners.",
    ),
    created_from: date | None = Query(
        None, description="Inclusive lower bound on created_at (YYYY-MM-DD).",
    ),
    created_to: date | None = Query(
        None, description="Inclusive upper bound on created_at (YYYY-MM-DD).",
    ),
    page: int = Query(1, ge=1),
    page_size: int = Query(50, ge=1, le=200),
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Public mail-group directory with filters.

    Any logged-in user can list. Filters are composable (AND). The
    ``only_mine`` flag is a UX shortcut mapping to ``owner_person_id``
    = current user; it wins over an explicit ``owner_person_id`` so
    the UI doesn't need to decide which to send.
    """
    effective_owner_id = owner_person_id
    if only_mine:
        effective_owner_id = current_user.id

    service = EmailGroupService(db)
    rows, total, persons_map = await service.list_groups(
        search=search,
        suffix=suffix,
        owner_person_id=effective_owner_id,
        only_unowned=only_unowned,
        created_from=_parse_day_bound(created_from, end_of_day=False),
        created_to=_parse_day_bound(created_to, end_of_day=True),
        page=page,
        page_size=page_size,
    )
    data = [_serialize_group(g, persons_map) for g in rows]
    return paginated_response(
        data=data, total=total, page=page, page_size=page_size, request=request,
    )


@router.get("/owners-directory")
async def list_mail_group_owners_directory(
    request: Request,
    _current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Distinct active owners across all mail groups, with name + email.

    Powers the owner-filter dropdown on the public directory so the
    "search by name or email" placeholder actually works. **No new PII
    surface**: the same ``work_email`` field is already returned for
    every owner of every group via the public ``GET /email-groups``
    listing (``EmailGroupOwnerResponse.work_email``). This endpoint is
    a search-convenience aggregation that lists ONLY people who
    already appear as an owner somewhere — it does NOT enumerate the
    full active-employee directory the way ``/auth/users`` does, so
    it's safe to expose to any authenticated user.

    Excludes leavers / left employees and persons soft-deleted on the
    Helix side; the picker's job is to filter the directory by an
    actionable owner.
    """
    # DISTINCT person_id across email_group_owners, joined to
    # IdentityPerson so we can return name + email in one round-trip
    # and skip leavers in SQL rather than after the fact.
    rows = (await db.execute(
        select(
            IdentityPerson.id,
            IdentityPerson.display_name,
            IdentityPerson.work_email,
        )
        .join(EmailGroupOwner, EmailGroupOwner.person_id == IdentityPerson.id)
        .where(
            IdentityPerson.deleted_at.is_(None),
            IdentityPerson.employment_status == "active",
        )
        .distinct()
        .order_by(
            # ``coalesce`` collapses NULL display_name to work_email so
            # name-less rows (rare but possible) sort by their email
            # rather than landing at the top of the list (MySQL's
            # default NULL-FIRST ordering for ASC).
            func.coalesce(IdentityPerson.display_name, IdentityPerson.work_email),
            IdentityPerson.id,
        )
    )).all()
    data = [
        EmailGroupOwnerDirectoryEntry(
            id=row.id,
            display_name=row.display_name or "",
            work_email=row.work_email or "",
        ).model_dump(mode="json")
        for row in rows
    ]
    return success_response(data=data, request=request)


# ── Writes ──

@router.post("/requests")
# Creating a mail group provisions a real Google Group. A duplicate
# submit leaves an orphaned group nobody asked for, so this one is
# keyed even though it is a creation flow rather than a state
# transition. Keyed on the requester: the same person re-submitting the
# SAME key is a retry, while a different person with a colliding key is
# a different operation.
@idempotent("email_group.create", key_params=("current_user.id",))
async def create_mail_group(
    request: Request,
    body: EmailGroupCreateRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    result = await service.create_mail_group(
        body=body,
        requester=current_user,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    # Reload the group so the returned payload has fresh owners
    persons_map: dict[UUID, IdentityPerson] = {}
    group = await service.find_group_by_email(result.group.email)
    if group is not None:
        if group.owners:
            owner_ids = [o.person_id for o in group.owners]
            persons = (
                await db.execute(
                    select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids))
                )
            ).scalars().all()
            persons_map = {p.id: p for p in persons}
    return success_response(
        data={
            "request_id": str(result.request.id),
            "request_number": result.request.request_number,
            "group": _serialize_group(group, persons_map) if group else None,
        },
        request=request,
    )


class EmailGroupDeleteBody(BaseModel):
    """Delete-with-reason payload.

    ``reason`` lands in the audit ``detail`` and requires at least 4
    characters — it forces the clicker to spell out *why* an
    irreversible Google delete is happening.

    (An earlier version of this docstring also promised "the Feishu
    notification to the group's remaining owners". There is no such
    notification — this domain ships ten notification templates and not
    one of them is a deletion. A reader trusting that line would
    believe owners are told their group is gone; nobody is. Recorded
    rather than quietly deleted, because the gap is real: the
    notification is worth adding, the false claim was worse than the
    gap.)
    """

    reason: str = Field(..., min_length=4, max_length=500)
    acknowledge_blocking: bool = Field(
        False,
        description=(
            "Delete even though the group carries BLOCKING signals "
            "(nested in another group, a real second address, a live "
            "BigQuery dataset grant, referenced in Helix config). Only "
            "HR / global admin may set it; the acknowledgement is "
            "recorded in the audit row."
        ),
    )


@router.delete("/{group_id}")
async def delete_mail_group(
    group_id: UUID,
    request: Request,
    body: EmailGroupDeleteBody,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    await service.delete_mail_group(
        group_id=group_id,
        actor=current_user,
        actor_roles=current_user.role_names,
        trace_id=request.state.trace_id,
        reason=body.reason,
        acknowledge_blocking=body.acknowledge_blocking,
    )
    await db.commit()
    return success_response(data={"deleted": True}, request=request)


@router.post("/{group_id}/transfer-owner")
async def transfer_group_owner_endpoint(
    group_id: UUID,
    request: Request,
    body: EmailGroupTransferOwnerRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """On-demand owner transfer (directory page).

    Moves a single owner seat from ``from_person_id`` → ``to_person_id``.
    Authorization: the caller must be either one of the group's current
    owners OR HR / IT Admin. Real Google calls happen — new owner added
    FIRST, then old removed, so Google's "at least one owner" invariant
    never trips. Reason is required (≥4 chars) and lands in audit + the
    new-owner notification body.
    """
    service = EmailGroupService(db)
    await service.transfer_owner_on_demand(
        group_id=group_id,
        from_person_id=body.from_person_id,
        to_person_id=body.to_person_id,
        actor=current_user,
        actor_roles=current_user.role_names,
        reason=body.reason,
        trace_id=request.state.trace_id,
    )
    await db.commit()

    # Re-load the group with owners eager-loaded so the returned
    # payload reflects the post-swap state. Doing a fresh SELECT
    # (instead of accessing ``group.owners`` on the pre-commit
    # instance) avoids a post-commit lazy-load inside a non-async
    # greenlet context.
    from sqlalchemy.orm import selectinload
    fresh = (
        await db.execute(
            select(EmailGroup)
            .where(EmailGroup.id == group_id)
            .options(selectinload(EmailGroup.owners))
        )
    ).scalar_one()
    persons_map: dict[UUID, IdentityPerson] = {}
    if fresh.owners:
        owner_ids = [o.person_id for o in fresh.owners]
        persons = (
            await db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(owner_ids))
            )
        ).scalars().all()
        persons_map = {p.id: p for p in persons}
    return success_response(
        data={"group": _serialize_group(fresh, persons_map)}, request=request,
    )


# ── Join requests (in-Helix owner approval) ──

@router.get(
    "/join-requests/pending",
    summary="Pending join requests the caller may decide (owner / admin)",
)
async def list_pending_join_requests(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Owner-scoped 待审批 queue.

    A plain owner sees pending requests on groups they own (minus their
    own); HR / IT Admin see every pending request. Data-driven duty —
    no role gate on the endpoint itself; the service scopes the rows.
    """
    service = EmailGroupService(db)
    person = await db.get(IdentityPerson, current_user.id)
    if person is None:
        raise NotFoundError("IdentityPerson", str(current_user.id))
    data = await service.list_pending_join_requests_for_owner(
        person=person, actor_roles=current_user.role_names or [],
    )
    return success_response(data=data, request=request)


@router.get(
    "/join-requests/mine",
    summary="Every join request the caller has submitted (any status)",
)
async def list_my_join_requests(
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    person = await db.get(IdentityPerson, current_user.id)
    if person is None:
        raise NotFoundError("IdentityPerson", str(current_user.id))
    data = await service.list_my_join_requests(requester=person)
    return success_response(data=data, request=request)


@router.post(
    "/{group_id}/join-requests",
    summary="Request to join a mail group (routes to the group's owner)",
)
async def create_join_request(
    group_id: UUID,
    request: Request,
    body: JoinGroupRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    req, has_owner = await service.request_join(
        group_id=group_id,
        requester=current_user,
        reason=body.reason,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    return success_response(
        data={
            "request_id": str(req.id),
            "status": req.status,
            # False → the group has no owner; the request was routed to
            # the HR / IT-Admin 待审批 queue, so the UI sets expectations.
            "has_owner": has_owner,
        },
        request=request,
    )


@router.post(
    "/join-requests/{request_id}/decide",
    summary="Owner / admin approves or rejects a join request",
)
async def decide_join_request(
    request_id: UUID,
    request: Request,
    body: JoinDecisionRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    req = await service.decide_join(
        request_id=request_id,
        actor=current_user,
        actor_roles=current_user.role_names or [],
        decision=body.decision,
        note=body.note,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    return success_response(
        data={"request_id": str(req.id), "status": req.status},
        request=request,
    )


@router.post(
    "/join-requests/{request_id}/cancel",
    summary="Requester withdraws their own pending join request",
)
async def cancel_join_request(
    request_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    req = await service.cancel_join(
        request_id=request_id,
        requester=current_user,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    return success_response(
        data={"request_id": str(req.id), "status": req.status},
        request=request,
    )


# ── Member management (owner / admin) ──

@router.get(
    "/{group_id}/members",
    summary="List a group's members (owner / admin)",
)
async def list_group_members_endpoint(
    group_id: UUID,
    request: Request,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    data = await service.list_members(
        group_id=group_id,
        actor=current_user,
        actor_roles=current_user.role_names or [],
    )
    return success_response(data=data, request=request)


@router.post(
    "/{group_id}/members/remove",
    summary="Remove a member from a group (owner / admin)",
)
async def remove_group_member_endpoint(
    group_id: UUID,
    request: Request,
    body: GroupMemberRemoveRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    data = await service.remove_member(
        group_id=group_id,
        member_email=body.member_email,
        actor=current_user,
        actor_roles=current_user.role_names or [],
        reason=body.reason,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    return success_response(data=data, request=request)


@router.post("/sync-now", summary="Trigger the Google mail-group sync immediately")
async def trigger_group_sync(
    request: Request,
    _admin: AuthUser = Depends(require_admin()),
    db: AsyncSession = Depends(get_db),
):
    """Admin-triggered mirror — reuses the ``google.group_sync`` handler so
    both the scheduled run and the manual run go through the same code
    path (one set of semantics, one summary shape)."""
    from app.scheduled_jobs.google_handlers import run_google_group_sync

    summary = await run_google_group_sync(db, config={})
    return success_response(data=summary, request=request)


@router.post("/aliases")
# Same reasoning as group creation — an alias is a Google-side object.
@idempotent("email_group.create_alias", key_params=("current_user.id",))
async def create_alias(
    request: Request,
    body: EmailAliasCreateRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    result = await service.create_alias(
        body=body,
        requester=current_user,
        requester_roles=current_user.role_names,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    return success_response(
        data={
            "request_id": str(result.request.id),
            "request_number": result.request.request_number,
            "alias_email": result.alias_email,
        },
        request=request,
    )


@router.post("/password-reset")
async def reset_own_password(
    request: Request,
    body: PasswordResetRequest,
    current_user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    service = EmailGroupService(db)
    result = await service.reset_own_password(
        body=body,
        requester=current_user,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    # NOTE: response body intentionally does NOT include the password.
    # Delivery goes via Feishu + email (notification outbox). Returning
    # it here would leak into API logs / browser history.
    return success_response(
        data={
            "request_id": str(result.request.id),
            "request_number": result.request.request_number,
            "delivered_via": ["feishu", "email"],
        },
        request=request,
    )


@router.post(
    "/password-reset/admin",
    dependencies=[Depends(require_admin())],
)
async def admin_reset_password(
    request: Request,
    body: PasswordResetAdminRequest,
    current_user: AuthUser = Depends(get_current_user),
    _rate: None = Depends(check_sensitive_endpoint_rate_limit),
    db: AsyncSession = Depends(get_db),
):
    """Admin password reset.

    The ``require_admin()`` dependency runs BEFORE this body, so a
    non-admin caller is 403'd before any DB lookup happens. The
    earlier shape did the lookup first and then handed the role check
    to the service layer, which let any authenticated user probe
    person UUIDs and distinguish "exists + active" from "404" via
    response code (a low-severity oracle, but fixable for free here).
    The service layer keeps its own ``has_role`` re-check as
    defense-in-depth.
    """
    service = EmailGroupService(db)
    target = await db.get(IdentityPerson, body.target_person_id)
    # Direct attribute access (NOT ``getattr(...,"deleted_at", None)``):
    # ``IdentityPerson.deleted_at`` is a real ORM column. The defensive
    # ``getattr`` form silently returns None on a future column rename
    # and degrades a soft-delete check to "always allow", which is
    # exactly the bug class memo'd in CLAUDE.md
    # (feedback_no_getattr_default_for_model_columns).
    if target is None or target.deleted_at is not None:
        # Reject password resets on soft-deleted / offboarded persons —
        # admin intent is ambiguous (re-activation?) and we don't want
        # an out-of-band password landing on a departed employee.
        raise NotFoundError("IdentityPerson", str(body.target_person_id))
    result = await service.admin_reset_password(
        target_person=target,
        actor=current_user,
        actor_roles=current_user.role_names,
        reason=body.reason,
        trace_id=request.state.trace_id,
    )
    await db.commit()
    return success_response(
        data={
            "request_id": str(result.request.id),
            "request_number": result.request.request_number,
            "target_person_id": str(target.id),
            "delivered_via": ["feishu", "email"],
        },
        request=request,
    )