from datetime import datetime
from uuid import UUID

from fastapi import APIRouter, Depends, Query, Request
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.schemas import AuditLogResponse
from app.audit.service import AuditService
from app.auth.models import AuthUser
from app.core.database import get_db
from app.core.dependencies import (
    check_directory_read_rate_limit,
    check_sensitive_endpoint_rate_limit,
    get_current_user,
    require_audit_access,
)
from app.core.exceptions import AppException, ForbiddenError, NotFoundError
from app.core.masking import redact_invoice_basis_amounts
from app.core.rbac import (
    ASSET_ADMIN_ROLE,
    AUDITOR_ROLE,
    CLOUD_ADMIN_ROLE,
    GLOBAL_ADMIN_ROLE,
    MANAGEMENT_ROLE,
    has_role,
)
from app.core.response import paginated_response, success_response
from app.identity.person_models import IdentityPerson

router = APIRouter(prefix="/api/v1/audit", tags=["audit"])


def _can_view_invoice(user: AuthUser) -> bool:
    """实付口径 clearance of the audit READER (finance / cloud_admin /
    global_admin).

    ``can_access_audit`` is wider than the invoice circle — it admits
    ``auditor`` and ``management`` — and cloud bill-alert audit rows
    carry the evaluated amounts in ``detail``. Every read path in this
    module resolves the reader's clearance here and hands it to
    ``_enrich_audit_logs`` / ``redact_invoice_basis_amounts``; the gate
    itself is deliberately NOT narrowed (an auditor seeing why an alert
    fired is their job — the amounts are the only confidential part).

    Lazy domain import: matches the ``CmdbNetworkOwnerHistory`` pattern
    below and keeps ``app.audit`` free of an import-time dependency on
    the CMDB model module.
    """
    from app.domains.cmdb.cost_view import can_view_invoice_measure
    return can_view_invoice_measure(user.role_names)


async def _enrich_audit_logs(
    db: AsyncSession,
    logs: list,
    *,
    can_view_invoice: bool = False,
) -> list[dict]:
    """Attach actor_display_name + actor_email via a single batch JOIN.

    Audit logs are queried in pages of ≤200, so a single ``IN (…)``
    lookup against identity_persons is cheap and avoids N+1 calls. The
    enrichment is intentionally kept on the read path (not denormalised
    onto audit_logs) so renames in the directory always surface — audit
    rows themselves stay immutable.

    ``can_view_invoice`` defaults to ``False`` so a caller that forgets
    to pass it (including the MCP audit tools, which reach into this
    helper) fails SAFE — over-redacted, never over-shared.
    """
    person_ids = {log.actor_user_id for log in logs if log.actor_user_id}
    person_map: dict[str, tuple[str | None, str | None]] = {}
    if person_ids:
        result = await db.execute(
            select(
                IdentityPerson.id,
                IdentityPerson.display_name,
                IdentityPerson.work_email,
                IdentityPerson.employee_no,
                IdentityPerson.employment_status,
            ).where(IdentityPerson.id.in_(person_ids))
        )
        person_map = {
            str(r.id): (r.display_name, r.work_email, r.employee_no, r.employment_status)
            for r in result
        }

    enriched: list[dict] = []
    for log in logs:
        d = AuditLogResponse.model_validate(log).model_dump(mode="json")
        d["detail"] = redact_invoice_basis_amounts(
            d.get("detail"),
            action=log.action,
            can_view_invoice=can_view_invoice,
        )
        if log.actor_user_id:
            person = person_map.get(str(log.actor_user_id))
            if person:
                (
                    d["actor_display_name"],
                    d["actor_email"],
                    d["actor_employee_no"],
                    d["actor_employment_status"],
                ) = person
        enriched.append(d)
    return enriched


@router.get("/stats")
async def get_audit_stats(
    request: Request,
    user: AuthUser = Depends(require_audit_access()),
    db: AsyncSession = Depends(get_db),
):
    service = AuditService(db)
    stats = await service.get_stats()
    return success_response(data=stats, request=request)


@router.get("/analytics")
async def get_audit_analytics(
    request: Request,
    window_days: int = 30,
    user: AuthUser = Depends(require_audit_access()),
    # Parity with ``/audit/logs``: the analytics surface is a heavier
    # aggregate (180-day daily GROUP BY across the largest write-mostly
    # table) AND it returns enriched ``top_actors[].actor_display_name +
    # actor_email``. A compromised auditor PAT could otherwise loop
    # this endpoint to (a) drain leaderboards, (b) chip activity
    # heat-maps, (c) burn worker DB. Same per-user directory-read
    # bucket (~600/h) as the list surface.
    _rl: None = Depends(check_directory_read_rate_limit),
    db: AsyncSession = Depends(get_db),
):
    """Aggregate dashboard payload for the audit-log 分析 tab.

    Surfaces daily / hourly trend, status pie, action / resource_type /
    actor leaderboards, action-prefix breakdown. Read-only — same
    auditor / admin gate as the rest of the audit surface.

    ``window_days`` is clamped to 1..180 — the audit table is the
    largest write-mostly surface on the platform, so a 180-day
    aggregate is the practical ceiling before the daily-by-day query
    plan starts to drag on TiDB. (The dashboard itself defaults to
    30d.)
    """
    # Explicit ``is None`` check (not falsy short-circuit) so an
    # explicit ``?window_days=0`` clamps to 1 instead of silently
    # resolving to the default 30 via ``0 or 30``.
    window_days = max(1, min(int(window_days if window_days is not None else 30), 180))
    service = AuditService(db)
    data = await service.get_analytics(window_days=window_days)
    return success_response(data=data, request=request)


@router.get("/actions")
async def get_audit_actions(
    request: Request,
    user: AuthUser = Depends(require_audit_access()),
    db: AsyncSession = Depends(get_db),
):
    service = AuditService(db)
    actions = await service.get_distinct_actions()
    return success_response(data=actions, request=request)


@router.get("/resource-types")
async def get_audit_resource_types(
    request: Request,
    user: AuthUser = Depends(require_audit_access()),
    db: AsyncSession = Depends(get_db),
):
    service = AuditService(db)
    resource_types = await service.get_distinct_resource_types()
    return success_response(data=resource_types, request=request)


@router.get("/export")
async def export_audit_logs(
    request: Request,
    action: str | None = None,
    actor_user_id: UUID | None = None,
    resource_type: str | None = None,
    resource_id: str | None = None,
    search: str | None = None,
    actor_type: str | None = None,
    status: str | None = None,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    # Row cap 5000 (not the global EXPORT_ROW_LIMIT 20000): audit
    # ``detail`` cells can carry up to ~16 KB of JSON each, so 5000
    # rows bounds the in-memory workbook at a manageable size while
    # still being 10× the old CSV cap.
    page_size: int = Query(default=5000, ge=1, le=5000),
    user: AuthUser = Depends(require_audit_access()),
    # Pre-launch hardening 2026-05-04: rate-limit the audit log export
    # so a Security Auditor account can't be used as a bulk-pull
    # exfiltration vehicle (and so a buggy admin script can't dump
    # the full table in a tight loop).
    _rl: None = Depends(check_sensitive_endpoint_rate_limit),
    db: AsyncSession = Depends(get_db),
):
    """Export the filtered audit log as XLSX via the unified export
    primitive (:mod:`app.core.exporting`). The audit action name stays
    the grandfathered ``audit.export``."""
    # Bulk-export of the entire audit table is a serious exfil vector
    # — require at least ONE scoping filter so an auditor / management
    # role can't iterate ``date_to`` and drain the table to S3. The
    # rate limit (10 calls/hour) is the second line of defence; this
    # is the first. Self-export of the actor_user_id is always
    # accepted (per-person history is a legitimate use).
    if (
        date_from is None
        and actor_user_id is None
        and resource_id is None
        and action is None
        and resource_type is None
        and search is None
    ):
        raise AppException(
            code="AUDIT_EXPORT_SCOPE_REQUIRED",
            message=(
                "Audit log export requires at least one scoping filter: "
                "date_from, actor_user_id, resource_type, resource_id, "
                "action, or search."
            ),
            status_code=400,
        )
    from app.core.exporting import xlsx_export_response

    service = AuditService(db)
    logs, total = await service.query_logs(
        page=1,
        page_size=page_size,
        action=action,
        actor_user_id=actor_user_id,
        resource_type=resource_type,
        resource_id=resource_id,
        search=search,
        actor_type=actor_type,
        status=status,
        date_from=date_from,
        date_to=date_to,
    )

    columns = [
        ("created_at", "时间"),
        ("action", "操作"),
        ("actor_type", "操作人类型"),
        ("actor_user_id", "操作人ID"),
        ("resource_type", "资源类型"),
        ("resource_id", "资源ID"),
        ("status", "状态"),
        ("ip_address", "IP"),
        ("trace_id", "Trace ID"),
        ("detail", "详情"),
        ("id", "ID"),
    ]
    # ``detail`` dicts are JSON-stringified by the primitive's
    # ``to_cell``; datetimes / UUIDs are coerced there too. The
    # invoice-basis redaction MUST run here too — an XLSX pull is the
    # same disclosure as the JSON list, only durable.
    export_can_view_invoice = _can_view_invoice(user)
    rows = [
        {
            "created_at": log.created_at,
            "action": log.action,
            "actor_type": log.actor_type,
            "actor_user_id": log.actor_user_id,
            "resource_type": log.resource_type,
            "resource_id": log.resource_id,
            "status": log.status,
            "ip_address": log.ip_address,
            "trace_id": log.trace_id,
            "detail": redact_invoice_basis_amounts(
                log.detail,
                action=log.action,
                can_view_invoice=export_can_view_invoice,
            ),
            "id": log.id,
        }
        for log in logs
    ]
    # The primitive writes the export audit row — the exporter's
    # identity and full filter set are durably recorded so a future
    # review can tell "who pulled what slice when".
    return await xlsx_export_response(
        db,
        request,
        user,
        columns=columns,
        rows=rows,
        total_matched=total,
        filename_stem="audit_logs",
        sheet_title="Audit Logs",
        audit_action="audit.export",
        audit_resource_type="audit_logs",
        filters={
            "action": action,
            "actor_user_id": actor_user_id,
            "resource_type": resource_type,
            "resource_id": resource_id,
            "search": (search or "")[:200] or None,
            "actor_type": actor_type,
            "status": status,
            "date_from": date_from,
            "date_to": date_to,
            "page_size": page_size,
        },
    )


@router.get("/logs")
async def query_audit_logs(
    request: Request,
    page: int = 1,
    page_size: int = Query(default=20, ge=1, le=200),
    action: str | None = None,
    actor_user_id: UUID | None = None,
    resource_type: str | None = None,
    resource_id: str | None = None,
    search: str | None = None,
    actor_type: str | None = None,
    status: str | None = None,
    date_from: datetime | None = None,
    date_to: datetime | None = None,
    user: AuthUser = Depends(require_audit_access()),
    # Apply the wider per-user directory-read bucket (~600/h) — same
    # exfil concern as ``/audit/export`` but the LIST surface drives
    # the interactive admin UI (page/filter/date changes burn calls
    # quickly during an incident). The narrower
    # ``check_sensitive_endpoint_rate_limit`` (10/h) is still applied
    # to ``/audit/export`` where bulk pulls are the only legitimate
    # mode.
    _rl: None = Depends(check_directory_read_rate_limit),
    db: AsyncSession = Depends(get_db),
):
    service = AuditService(db)
    logs, total = await service.query_logs(
        page=page,
        page_size=page_size,
        action=action,
        actor_user_id=actor_user_id,
        resource_type=resource_type,
        resource_id=resource_id,
        search=search,
        actor_type=actor_type,
        status=status,
        date_from=date_from,
        date_to=date_to,
    )
    data = await _enrich_audit_logs(
        db, logs, can_view_invoice=_can_view_invoice(user),
    )
    return paginated_response(data=data, total=total, page=page, page_size=page_size, request=request)


# ── Per-resource history ──────────────────────────────────────────────
#
# ``/audit/logs`` is the privileged-auditor swiss army surface — gated
# on ``can_access_audit`` (admin / auditor / management) because it
# leaks the full action ledger across the platform: who reset whose
# password, which integration credentials moved, every leaver
# blocker, etc. Surfacing it from a DNS detail page would require
# every Cloud Admin to also hold auditor.
#
# ``/audit/resources/history`` is a narrower lens: "every audit row
# attached to this one resource." The access check is resource-type
# aware so a Cloud Admin sees DNS / domain / server history without
# needing auditor, an Asset Admin sees asset history without needing
# auditor, etc. Cross-resource fishing is impossible because the
# endpoint requires both ``resource_type`` and ``resource_id`` and
# returns rows for that single id only — there is no "list all" path.

# Resource-type → allowed-role-set mapping for the per-resource
# history view. Keys are matched as prefixes so the seven CMDB
# resource types share a single rule. Anything not matched falls
# back to ``can_access_audit`` (admin / auditor / management) so
# new domains never accidentally over-share before the table here
# is updated.
_CLOUD_HISTORY_ROLES = (
    GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, CLOUD_ADMIN_ROLE, MANAGEMENT_ROLE,
)
_ASSET_HISTORY_ROLES = (
    GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, ASSET_ADMIN_ROLE, MANAGEMENT_ROLE,
)
_PERMISSION_HISTORY_ROLES = (
    GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE,
)
# NOTE — ``email_group`` deliberately has NO entry below, so it falls
# through to the conservative audit-admin default.
#
# It briefly had one that added ``HR_ROLE``, to give the 邮件组管理 desk
# a per-group timeline. That was wrong: this endpoint filters by
# ``resource_type`` + ``resource_id`` with NO action filter, and the
# data-access domain writes ``data_access.binding.*`` rows against
# ``resource_type='email_group'`` (see ``data_access/service.py``
# ~4260 / ~5484 / ~5581). Opening it to HR therefore let anyone with
# the HR role walk group ids and read BigQuery dataset topology and
# revoke history — a different domain's governance data, reached
# through a mail-group screen.
#
# The desk gets its timeline from ``GET /email-groups/manage/audit-log``
# with ``group_id`` instead, which is constrained to
# ``action LIKE 'email_group.%'`` at the query level and gated by
# ``can_manage_email_groups``. Same information, none of the reach.

_RESOURCE_HISTORY_PREFIX_ACCESS: tuple[tuple[str, tuple[str, ...]], ...] = (
    # CMDB / cloud-side resources — Cloud Admin needs the lifecycle
    # view of "this DNS record" without auditor authority.
    ("cmdb", _CLOUD_HISTORY_ROLES),
    ("integration", _CLOUD_HISTORY_ROLES),
    # Asset domain — Asset Admin owns the laptop / equipment lifecycle.
    ("asset", _ASSET_HISTORY_ROLES),
    ("procurement_order", _ASSET_HISTORY_ROLES),
    ("employee_request", _ASSET_HISTORY_ROLES),
    # Software permissions / SSO grants — admin-only governance.
    ("user_permission", _PERMISSION_HISTORY_ROLES),
    ("software_permission", _PERMISSION_HISTORY_ROLES),
    ("sso_grant", _PERMISSION_HISTORY_ROLES),
)


# Cloud resource types whose owner changes also live in the dedicated
# ``cmdb_network_owner_history`` table (separate from ``audit_logs``).
# Reading just audit_logs would silently miss every owner transfer for
# these resources — the full lifecycle view must union the two. Maps
# the audit-side ``cmdb_*`` resource_type to the bare-name string the
# owner_history table uses (``RESOURCE_TYPE_*`` constants in
# [owner_history.py](../domains/cmdb/owner_history.py)). ``cloud_server``
# joined the enum in Round 45 (CR-205) and gained writers on every
# owner-mutation path (admin PATCH / batch / self-service transfer /
# offboarding force-transfer) — but this map was never updated, so the
# server timeline silently dropped every owner transfer until 2026-07-21.
_AUDIT_TO_OWNER_HISTORY_TYPE: dict[str, str] = {
    "cmdb_dns_record": "dns_record",
    "cmdb_domain": "domain",
    "cmdb_cdn_domain": "cdn_domain",
    "cmdb_load_balancer": "load_balancer",
    "cmdb_server": "cloud_server",
}


def _allowed_roles_for_resource_type(resource_type: str) -> tuple[str, ...]:
    """Return the role-set permitted to read history for ``resource_type``.

    Match is exact-or-underscore-prefix so ``asset`` matches both the
    bare ``asset`` type AND ``asset_supplier`` / ``asset_business_payment``
    / etc., but does NOT accidentally match an unrelated future type
    that happens to share leading letters (``assetextra``). Default is
    the conservative audit-admin gate when the type is unknown — every
    new domain must be listed explicitly above before its history can
    be surfaced to non-auditors.
    """
    for prefix, roles in _RESOURCE_HISTORY_PREFIX_ACCESS:
        if resource_type == prefix or resource_type.startswith(prefix + "_"):
            return roles
    return (GLOBAL_ADMIN_ROLE, AUDITOR_ROLE, MANAGEMENT_ROLE)


async def _fetch_owner_history_rows(
    db: AsyncSession,
    *,
    owner_history_types: list[str],
    resource_id: str,
    can_view_invoice: bool = False,
) -> list[dict]:
    """Pull every ``cmdb_network_owner_history`` row for the resource and
    re-shape them as audit-log-flavoured dicts so the per-resource
    timeline can merge them with ``audit_logs`` in a single chronological
    list.

    Owner history is bounded per resource — even a chatty DNS record
    sees ~10 owner changes over its life — so we fetch all of them
    unconditionally rather than trying to paginate across two tables
    (correct pagination would require an interleaved SQL UNION which
    MySQL/TiDB will scan inefficiently when the secondary table is
    tiny). The merge then paginates in Python on the combined sorted
    set.
    """
    if not owner_history_types:
        return []
    from app.domains.cmdb.models import CmdbNetworkOwnerHistory

    rows = (await db.execute(
        select(CmdbNetworkOwnerHistory)
        .where(
            CmdbNetworkOwnerHistory.resource_type.in_(owner_history_types),
            CmdbNetworkOwnerHistory.resource_id == resource_id,
        )
        .order_by(CmdbNetworkOwnerHistory.created_at.desc())
    )).scalars().all()

    # Batch-resolve person labels: actor (writer of the change) becomes
    # the audit row's actor; previous + new owner_person_id end up in
    # ``detail`` so the row reads "X transferred ownership Y → Z" at a
    # glance. ``context`` already snapshots ``previous_label`` /
    # ``new_label`` per migration 175 + Round-28, so the forensic value
    # survives a future hard-delete of the person FK target.
    person_ids: set[str] = set()
    for row in rows:
        if row.actor_person_id:
            person_ids.add(str(row.actor_person_id))
    person_map: dict[str, tuple[str | None, str | None, str | None, str | None]] = {}
    if person_ids:
        result = await db.execute(
            select(
                IdentityPerson.id,
                IdentityPerson.display_name,
                IdentityPerson.work_email,
                IdentityPerson.employee_no,
                IdentityPerson.employment_status,
            ).where(IdentityPerson.id.in_(person_ids))
        )
        person_map = {
            str(r.id): (r.display_name, r.work_email, r.employee_no, r.employment_status)
            for r in result
        }

    # Map the bare owner_history type back to the canonical audit
    # ``cmdb_*`` string so the merged timeline's ``resource_type``
    # column stays consistent for the consumer (the row badge in the
    # drawer reads "cmdb_dns_record" not the schema-detail "dns_record").
    type_back_map = {v: k for k, v in _AUDIT_TO_OWNER_HISTORY_TYPE.items()}
    out: list[dict] = []
    for row in rows:
        actor_id = str(row.actor_person_id) if row.actor_person_id else None
        display, email, emp_no, emp_status = (
            person_map.get(actor_id, (None, None, None, None))
            if actor_id
            else (None, None, None, None)
        )
        ctx = row.context if isinstance(row.context, dict) else {}
        previous_label = ctx.get("previous_label")
        new_label = ctx.get("new_label")
        # ``context`` is free-form per writer, so it is exactly the kind
        # of blob that grows a money key nobody re-reviews. Run it
        # through the same redactor under the synthetic action name —
        # unclassified ⇒ fail safe.
        synthetic_action = f"cmdb.owner_history.{row.change_reason}"
        history_detail = redact_invoice_basis_amounts(
            {
                "change_reason": row.change_reason,
                "previous_owner_person_id": row.previous_owner_person_id,
                "new_owner_person_id": row.new_owner_person_id,
                "previous_owner_label": previous_label,
                "new_owner_label": new_label,
                **{
                    k: v for k, v in ctx.items()
                    if k not in {"previous_label", "new_label"}
                },
            },
            action=synthetic_action,
            can_view_invoice=can_view_invoice,
        )
        out.append({
            "id": str(row.id),
            # ``cmdb_network_owner_history.trace_id`` is ``str | None``
            # (free-form 40-char column, not UUID-typed). Pass through
            # without coercion — the AuditLog response shape declares
            # ``trace_id: UUID`` but the merged endpoint does NOT
            # validate synth rows through the schema (this dict is
            # returned via ``paginated_response`` directly). Emitting
            # None when absent is cleaner than an empty string both
            # for the JSON consumer (a falsy value the drawer can
            # render "—") and for any future schema tightening.
            "trace_id": row.trace_id or None,
            "actor_user_id": actor_id,
            "actor_type": "user" if actor_id else "system",
            "actor_display_name": display,
            "actor_email": email,
            "actor_employee_no": emp_no,
            "actor_employment_status": emp_status,
            # Synthetic action namespace ``cmdb.owner_history.*`` — keeps
            # the action tag distinct from real audit_logs writes so a
            # forensic reader can tell the row was derived from
            # owner_history (no IP / UA / detail JSON) versus a direct
            # audit emission.
            "action": synthetic_action,
            "resource_type": type_back_map.get(row.resource_type, row.resource_type),
            "resource_id": row.resource_id,
            "status": "success",
            "detail": history_detail,
            "ip_address": None,
            "user_agent": None,
            "created_at": row.created_at.isoformat() if row.created_at else None,
        })
    return out


@router.get("/resources/history")
async def get_resource_history(
    request: Request,
    resource_type: str = Query(
        ...,
        description=(
            "Audit ``resource_type`` to filter on. Comma-separated to "
            "request a multi-type IN-list filter."
        ),
        min_length=1,
        max_length=512,
    ),
    resource_id: str = Query(
        ...,
        description="Exact audit resource_id (UUID or domain-scoped id).",
        min_length=1,
        max_length=128,
    ),
    page: int = 1,
    page_size: int = Query(default=20, ge=1, le=200),
    user: AuthUser = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Return the full lifecycle timeline for a single resource.

    For cloud network resources (DNS / domain / CDN / LB) the timeline
    unions ``audit_logs`` rows with the dedicated
    ``cmdb_network_owner_history`` table — owner transfers are NOT
    written into ``audit_logs`` by the lifecycle service, so reading
    just audit_logs would hide every "X → Y" transfer that the admin
    care most about. The two streams are merged client-side-of-the-DB
    in Python (owner history is bounded per resource), sorted by
    created_at desc, and paginated against the combined total.

    Access is gated per resource_type — a Cloud Admin sees DNS / domain
    / server history; an Asset Admin sees asset history; etc. Every
    requested resource_type must be allowed; mismatched callers receive
    ``403`` and are NOT told which type was the offender (avoid the
    "probe by mixing types" anti-pattern). The endpoint is read-only
    so there is no per-row mutation surface to defend.
    """
    rt_list = [s.strip() for s in resource_type.split(",") if s.strip()]
    if not rt_list:
        # Empty / whitespace-only after the split. The FastAPI
        # ``Query(..., min_length=1)`` already rejects an empty raw
        # query param; this guard catches the comma-only / spaces-only
        # case. Return 400 (client error in the QUERY, not authz),
        # matching the semantics of every other "this param is empty"
        # path in the router.
        from app.core.exceptions import ValidationError as _ValidationError
        raise _ValidationError("resource_type is required.")
    role_names = list(user.role_names)
    for rt in rt_list:
        allowed = _allowed_roles_for_resource_type(rt)
        if not has_role(role_names, *allowed):
            # Generic 403 — do not echo back which resource_type failed
            # so a caller cannot enumerate access boundaries by probing
            # different types.
            raise ForbiddenError(
                "You do not have access to this resource's history.",
            )

    # Derive matching owner_history row-type list (subset of rt_list
    # that have an owner-history mirror). Empty list means no merge.
    owner_history_types = [
        _AUDIT_TO_OWNER_HISTORY_TYPE[rt]
        for rt in rt_list
        if rt in _AUDIT_TO_OWNER_HISTORY_TYPE
    ]

    service = AuditService(db)
    # ``/resources/history`` widens the audience further than
    # ``/audit/logs``: a ``cmdb*`` resource_type admits cloud_admin (in
    # the invoice circle) AND auditor / management (outside it). Same
    # per-reader redaction, resolved once for both streams.
    can_view_invoice = _can_view_invoice(user)

    # Owner history fetched in full (bounded per resource — even a
    # heavily-recertified DNS record sees a few dozen owner changes
    # over its life, indexed by composite resource_type+resource_id).
    owner_dicts = await _fetch_owner_history_rows(
        db,
        owner_history_types=owner_history_types,
        resource_id=resource_id,
        can_view_invoice=can_view_invoice,
    )

    # Audit-logs side: pull enough audit rows from the top of the
    # date-desc stream to satisfy the requested page after merge.
    # Worst case the page is filled entirely by audit rows (every
    # owner_history row sorts after page P) — so we need at least
    # ``page * page_size`` audit rows; +K extra to interleave with
    # owner_history rows that might fall into the page window.
    # Composite index ix_audit_logs_resource_lookup keeps this O(N)
    # in the offset, not a full scan.
    #
    # Hard upper bound ``audit_prefetch_cap`` = 10 000 audit rows
    # (= 500 pages of 20). Resources with more than 10 000 audit
    # rows are pathological — a single resource gathering that many
    # state changes points to a misbehaving cron / driver, not normal
    # use — and the global ``/audit/logs`` search remains available
    # as the firehose for forensic deep-dives. Once the prefetch is
    # at the cap, ``total`` is reported truthfully from the indexed
    # COUNT(*), so the UI can tell the user "<N> entries (showing
    # latest <cap>)" instead of silently lying.
    audit_prefetch_cap = 10_000
    k = len(owner_dicts)
    audit_rows_needed = min(audit_prefetch_cap, page * page_size + k)
    audit_rows, audit_total = await service.query_logs(
        page=1,
        page_size=audit_rows_needed,
        resource_type=resource_type,
        resource_id=resource_id,
    )
    audit_dicts = await _enrich_audit_logs(
        db, audit_rows, can_view_invoice=can_view_invoice,
    )

    # Merge + sort by created_at desc. Both producers emit naive ISO
    # timestamps from the same source clock (DB UTC) — audit_logs
    # through Pydantic ``model_dump(mode='json')`` and owner_history
    # through ``datetime.isoformat()``. On MySQL/TiDB the
    # ``DateTime(timezone=True)`` column with asyncmy returns naive
    # values, so both producers emit
    # ``YYYY-MM-DDTHH:MM:SS[.ffffff]`` with NO timezone suffix and
    # lexicographic comparison IS chronological. If we ever migrate
    # to a TZ-aware datetime backend both producers MUST consistently
    # emit the same suffix (e.g. trailing ``Z``) or this sort skews —
    # add a normaliser at the producer layer in that future.
    merged = audit_dicts + owner_dicts
    merged.sort(
        key=lambda r: r.get("created_at") or "",
        reverse=True,
    )

    # The honest total: every audit row that matched (NOT capped by
    # our prefetch — ``audit_total`` is the COUNT(*) from the indexed
    # WHERE filter) + every owner_history row we fetched. UI uses
    # this to draw pagination. If ``total`` exceeds the prefetch
    # window the trailing pages return empty (the data exists but
    # this endpoint refuses to scan past the safety cap); admins who
    # really need page 500+ go to ``/admin/audit`` with the
    # resource_id filter.
    total = audit_total + k
    start = (page - 1) * page_size
    end = start + page_size
    return paginated_response(
        data=merged[start:end],
        total=total,
        page=page,
        page_size=page_size,
        request=request,
    )


@router.get("/logs/{log_id}")
async def get_audit_log_detail(
    log_id: UUID,
    request: Request,
    user: AuthUser = Depends(require_audit_access()),
    db: AsyncSession = Depends(get_db),
):
    service = AuditService(db)
    log = await service.get_by_id(log_id)
    if not log:
        raise NotFoundError("AuditLog", str(log_id))
    enriched = await _enrich_audit_logs(
        db, [log], can_view_invoice=_can_view_invoice(user),
    )
    return success_response(data=enriched[0], request=request)


@router.get("/trace/{trace_id}")
async def get_trace(
    trace_id: str,
    request: Request,
    user: AuthUser = Depends(require_audit_access()),
    db: AsyncSession = Depends(get_db),
):
    service = AuditService(db)
    logs = await service.get_by_trace_id(trace_id)
    data = await _enrich_audit_logs(
        db, logs, can_view_invoice=_can_view_invoice(user),
    )
    return success_response(data=data, request=request)