from collections.abc import Awaitable, Callable
from datetime import UTC, datetime
from typing import Any
from uuid import NAMESPACE_OID, UUID, uuid5

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

from app.audit.models import AuditLog
from app.core.safe_logging import get_safe_logger

logger = get_safe_logger(__name__)


# ── Post-log hooks ──────────────────────────────────────────────────
# Generic, in-process extension point: callables registered here are
# invoked once per successful ``AuditService.log`` write, AFTER the
# audit row has been flushed to the caller's session (so it rides the
# same transaction). A hook receives ``(entry, session)`` and runs
# in the caller's async context, so it MAY await DB work / dispatch
# notifications — unlike the SQLAlchemy ``after_flush`` listeners
# (see ``app.identity.audit_listener``) which can only queue more rows.
#
# Contract: hooks are best-effort. A hook that raises is logged and
# swallowed — an observer must never break the audit write or the
# operation it audits. Register at startup (both API + worker
# processes) so the hook fires regardless of which process logged.
#
# Current consumers:
#   * ``app.domains.asset.oversight_notifier`` — pushes to the global
#     IT asset admin when a regional asset admin operates on an asset.
AuditLogHook = Callable[[AuditLog, AsyncSession], Awaitable[None]]
_AUDIT_LOG_HOOKS: list[AuditLogHook] = []


def register_audit_log_hook(hook: AuditLogHook) -> None:
    """Idempotently register a post-log hook. See ``_AUDIT_LOG_HOOKS``."""
    if hook not in _AUDIT_LOG_HOOKS:
        _AUDIT_LOG_HOOKS.append(hook)


def _esc(s: str) -> str:
    """Escape LIKE metacharacters to prevent wildcard injection."""
    return s.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


async def log_denied_action(
    *,
    trace_id: str | UUID,
    actor_type: str,
    action: str,
    actor_user_id: UUID | None = None,
    resource_type: str | None = None,
    resource_id: str | None = None,
    detail: dict[str, Any] | None = None,
    ip_address: str | None = None,
    user_agent: str | None = None,
) -> None:
    """Write a ``status='failure'`` audit row in an INDEPENDENT session
    and commit it before returning. Use this on every "denied / failed"
    code path that will then ``raise`` an HTTP exception.

    The standard ``AuditService(db).log(...)`` only flushes into the
    caller's session; if the caller subsequently raises, ``get_db``
    rolls everything back and the audit row is lost. Defeats the
    purpose of the security control. ``log_denied_action`` opens a
    fresh session, commits, and closes — same posture as the
    rate-limit Redis primitive — so the audit trail survives the
    request rollback.

    Best-effort: any failure to write / commit is logged but never
    re-raised; we MUST NOT mask the original ``ForbiddenError`` /
    ``NotFoundError`` etc. with an audit-side failure.
    """
    from app.core.database import async_session_factory

    try:
        async with async_session_factory() as new_session:
            try:
                local = AuditService(new_session)
                await local.log(
                    trace_id=trace_id,
                    actor_type=actor_type,
                    action=action,
                    status="failure",
                    actor_user_id=actor_user_id,
                    resource_type=resource_type,
                    resource_id=resource_id,
                    detail=detail,
                    ip_address=ip_address,
                    user_agent=user_agent,
                )
                await new_session.commit()
            except Exception:
                await new_session.rollback()
                raise
    except Exception:
        logger.exception(
            "audit.log_denied_action: failed to persist denied "
            "audit row (action=%s resource_id=%s) — request raise "
            "proceeds without an audit trail",
            action, resource_id,
        )


class AuditService:
    def __init__(self, db: AsyncSession):
        self.db = db

    @staticmethod
    def _coerce_trace_id(trace_id: str | UUID) -> UUID:
        """Accept UUID, UUID-shape string, or a free-form tag.

        Tags like ``"process_advance"`` (used by scheduled-job handlers
        that don't have an HTTP request's real trace_id) are hashed
        deterministically into a UUID v5 so the audit column stays
        valid and a consistent pseudo-trace clusters related actions.
        Empty / None falls back to a fresh UUID so the DB write never
        fails on an optional field.
        """
        if isinstance(trace_id, UUID):
            return trace_id
        s = str(trace_id or "").strip()
        if not s:
            from uuid import uuid4
            return uuid4()
        try:
            return UUID(s)
        except ValueError:
            return uuid5(NAMESPACE_OID, s)

    async def log(
        self,
        trace_id: str | UUID,
        actor_type: str,
        action: str,
        status: str,
        actor_user_id: UUID | None = None,
        resource_type: str | None = None,
        resource_id: str | None = None,
        detail: dict | None = None,
        ip_address: str | None = None,
        user_agent: str | None = None,
    ) -> AuditLog:
        # Merge per-request auth metadata (e.g. ``via=pat`` +
        # ``pat_token_id``) into ``detail`` so post-hoc forensics can
        # tell whether an action was taken via the web UI (JWT) or via
        # an MCP / API client (PAT) without every call site threading
        # the request object. Background jobs without a request set
        # nothing, so the merge is a no-op there.
        from app.auth.auth_context import auth_metadata
        meta = auth_metadata()
        if meta:
            merged = dict(detail) if detail else {}
            for k, v in meta.items():
                merged.setdefault(k, v)
            detail = merged

        # Pre-launch audit Round-29 fix: cap user_agent at 200 chars so
        # we (a) bound the audit row size — UA strings can run 500+ chars
        # on mobile / enterprise endpoints and bloat the audit_logs table
        # over time — and (b) avoid copying long device-fingerprint-like
        # suffixes (hardware model, app build SHA) into forensic logs.
        # The first ~200 chars carry the family + version we actually
        # need for "was this Chrome 134 or curl" attribution.
        ua_capped = user_agent[:200] if user_agent else user_agent

        # Round-30: cap detail JSON serialization at 16 KiB. Audit
        # callers occasionally pass nested provider response bodies or
        # user-supplied remark fields (notes, reasons) that can grow to
        # multi-MB; left unchecked these bloat the audit_logs table,
        # slow row-based replication, and break the CSV export which
        # truncates at 4KB per cell anyway (admit the cap upstream).
        # Replace oversized detail with a marker so forensic walks can
        # still tell "something happened" but the row stays bounded.
        if detail is not None:
            import json as _json
            try:
                serialized = _json.dumps(detail, default=str)
            except (TypeError, ValueError):
                detail = {
                    "_truncated": "audit.detail not JSON-serializable",
                }
            else:
                if len(serialized) > 16384:
                    detail = {
                        "_truncated": (
                            f"audit.detail exceeded 16KB "
                            f"({len(serialized)} bytes)"
                        ),
                        "_action": action,
                    }

        # 2026-05-26 defence-in-depth: ``audit_logs.resource_id`` is
        # ``String(128)`` (see [audit/models.py:25]). Callers
        # historically formatted long composite ids (``f"{integ}:{rid}"``
        # for cmdb resource-creator audit rows, Azure ARM ids, GCP
        # full-resource paths) and trusted the column to fit. TiDB
        # STRICT_TRANS_TABLES raises DataError on overflow → audit
        # write fails AFTER the upstream side-effect (e.g. CloudTrail
        # API quota burn) already happened. We truncate here so the
        # audit row always lands; truncation is forensics-acceptable
        # because (a) every site already prefixes a UUID that uniquely
        # identifies the row, (b) ``detail`` carries the unredacted
        # context, (c) the trailing tail of an ARM id is the most
        # specific portion and that's what we keep by slicing from
        # the front when callers exceed.
        resource_id_capped = (
            resource_id[:128] if resource_id else resource_id
        )

        # Same defence-in-depth as resource_id/user_agent above:
        # ``audit_logs.action`` is ``String(64)`` (see [audit/models.py]).
        # All current callers use bounded server-controlled enums, but a
        # future long ``action=f"..."`` would raise DataError under TiDB
        # STRICT *after* the side-effect already happened. Cap from the
        # front so the audit row always lands.
        action_capped = action[:64] if action else action

        entry = AuditLog(
            trace_id=self._coerce_trace_id(trace_id),
            actor_user_id=actor_user_id,
            actor_type=actor_type,
            action=action_capped,
            resource_type=resource_type,
            resource_id=resource_id_capped,
            status=status,
            detail=detail,
            ip_address=ip_address,
            user_agent=ua_capped,
        )
        self.db.add(entry)
        await self.db.flush()

        # Fan out to post-log hooks (best-effort). A hook runs in the
        # caller's transaction and may await — an observer failure must
        # never break the audit write or the audited operation.
        for hook in _AUDIT_LOG_HOOKS:
            try:
                await hook(entry, self.db)
            except Exception:  # noqa: BLE001
                logger.exception(
                    "audit post-log hook %r failed (action=%s "
                    "resource_id=%s) — swallowed",
                    getattr(hook, "__qualname__", hook),
                    entry.action,
                    entry.resource_id,
                )

        return entry

    async def query_logs(
        self,
        page: int = 1,
        page_size: int = 20,
        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,
    ) -> tuple[list[AuditLog], int]:
        query = select(AuditLog)
        count_query = select(func.count()).select_from(AuditLog)

        if action:
            query = query.where(AuditLog.action == action)
            count_query = count_query.where(AuditLog.action == action)
        if actor_user_id:
            query = query.where(AuditLog.actor_user_id == actor_user_id)
            count_query = count_query.where(AuditLog.actor_user_id == actor_user_id)
        if resource_type:
            # Accept comma-separated resource_type so the per-resource
            # history surface can collapse the codebase's existing
            # duplicate audit strings into a single timeline view —
            # DNS records, for example, are audited under BOTH
            # ``cmdb_dns_record`` (router/service) and ``dns_record``
            # (owner_history constants). A single-equality filter would
            # silently drop half the rows and the "lifecycle" view
            # would lie. Composite index ix_audit_logs_resource_lookup
            # (migration 180) covers leading ``resource_type`` so the
            # IN-list still range-scans, not full-scans.
            rt_list = [s.strip() for s in resource_type.split(",") if s.strip()]
            if len(rt_list) == 1:
                query = query.where(AuditLog.resource_type == rt_list[0])
                count_query = count_query.where(AuditLog.resource_type == rt_list[0])
            elif rt_list:
                query = query.where(AuditLog.resource_type.in_(rt_list))
                count_query = count_query.where(AuditLog.resource_type.in_(rt_list))
        if resource_id:
            # Exact match — the per-resource history surface needs
            # "every audit row for THIS specific id". Free-text
            # ``search`` already matches resource_id substrings, but
            # substring matching pulls neighbour ids (UUID prefix
            # collisions are rare but real) and is index-hostile;
            # equality on ``resource_id`` rides the composite index.
            query = query.where(AuditLog.resource_id == resource_id)
            count_query = count_query.where(AuditLog.resource_id == resource_id)
        if actor_type:
            query = query.where(AuditLog.actor_type == actor_type)
            count_query = count_query.where(AuditLog.actor_type == actor_type)
        if status:
            query = query.where(AuditLog.status == status)
            count_query = count_query.where(AuditLog.status == status)
        if search:
            # Use SQLAlchemy String type (not Python builtin str) for CAST.
            # Also parameterize via ilike bind param — pymysql escapes quotes
            # and other SQL special characters, so literal `' OR 1=1` in the
            # search term is stored as a literal substring, not SQL.
            search_filter = or_(
                AuditLog.trace_id.cast(String).ilike(f"%{_esc(search)}%"),
                AuditLog.resource_id.ilike(f"%{_esc(search)}%"),
                AuditLog.action.ilike(f"%{_esc(search)}%"),
            )
            query = query.where(search_filter)
            count_query = count_query.where(search_filter)
        if date_from:
            query = query.where(AuditLog.created_at >= date_from)
            count_query = count_query.where(AuditLog.created_at >= date_from)
        if date_to:
            query = query.where(AuditLog.created_at <= date_to)
            count_query = count_query.where(AuditLog.created_at <= date_to)

        total_result = await self.db.execute(count_query)
        total = total_result.scalar() or 0

        query = query.order_by(AuditLog.created_at.desc()).offset((page - 1) * page_size).limit(page_size)
        result = await self.db.execute(query)
        logs = list(result.scalars().all())

        return logs, total

    async def get_by_id(self, log_id: UUID) -> AuditLog | None:
        result = await self.db.execute(
            select(AuditLog).where(AuditLog.id == log_id)
        )
        return result.scalar_one_or_none()

    async def get_by_trace_id(self, trace_id: str) -> list[AuditLog]:
        try:
            trace_uuid = self._coerce_trace_id(trace_id)
        except (TypeError, ValueError):
            return []
        result = await self.db.execute(
            select(AuditLog)
            .where(AuditLog.trace_id == trace_uuid)
            .order_by(AuditLog.created_at.asc())
        )
        return list(result.scalars().all())

    async def get_stats(self) -> dict:
        # Total logs
        total_result = await self.db.execute(
            select(func.count()).select_from(AuditLog)
        )
        total_logs = total_result.scalar() or 0

        # Logs today
        today_start = datetime.now(UTC).replace(hour=0, minute=0, second=0, microsecond=0)
        today_result = await self.db.execute(
            select(func.count()).select_from(AuditLog).where(
                AuditLog.created_at >= today_start
            )
        )
        logs_today = today_result.scalar() or 0

        # Unique actors
        actors_result = await self.db.execute(
            select(func.count(func.distinct(AuditLog.actor_user_id)))
        )
        unique_actors = actors_result.scalar() or 0

        # By action (top 20)
        action_result = await self.db.execute(
            select(AuditLog.action, func.count().label("count"))
            .group_by(AuditLog.action)
            .order_by(func.count().desc())
            .limit(20)
        )
        by_action = [
            {"action": row.action, "count": row.count}
            for row in action_result.all()
        ]

        # By resource_type
        resource_result = await self.db.execute(
            select(AuditLog.resource_type, func.count().label("count"))
            .where(AuditLog.resource_type.isnot(None))
            .group_by(AuditLog.resource_type)
            .order_by(func.count().desc())
            .limit(20)
        )
        by_resource_type = [
            {"resource_type": row.resource_type, "count": row.count}
            for row in resource_result.all()
        ]

        # By status
        status_result = await self.db.execute(
            select(AuditLog.status, func.count().label("count"))
            .group_by(AuditLog.status)
            .order_by(func.count().desc())
        )
        by_status = [
            {"status": row.status, "count": row.count}
            for row in status_result.all()
        ]

        # Recent activity (last 10)
        recent_result = await self.db.execute(
            select(AuditLog)
            .order_by(AuditLog.created_at.desc())
            .limit(10)
        )
        recent_activity = [
            {
                "id": str(log.id),
                "trace_id": str(log.trace_id),
                "action": log.action,
                "resource_type": log.resource_type,
                "status": log.status,
                "created_at": log.created_at.isoformat(),
            }
            for log in recent_result.scalars().all()
        ]

        return {
            "total_logs": total_logs,
            "logs_today": logs_today,
            "unique_actors": unique_actors,
            "by_action": by_action,
            "by_resource_type": by_resource_type,
            "by_status": by_status,
            "recent_activity": recent_activity,
        }

    async def get_analytics(self, *, window_days: int = 30) -> dict:
        """Rich aggregates for the audit-log 分析 tab.

        Surfaces daily / hourly trend, status breakdown, top actions,
        top resource types, top human actors, and a user-vs-system
        split. Every aggregate runs against the existing audit_logs
        table indexes (``created_at`` desc + the resource composite).

        Window applies to the timeseries and "top actor" rollups —
        lifetime fields (``total_logs``, ``unique_actors``) stay
        snapshot-style to match the value an operator expects from a
        compliance ledger.
        """

        from sqlalchemy import case

        from app.core.timeseries import day_axis
        from app.identity.person_models import IdentityPerson

        now = datetime.now(UTC)
        # Round-62 P1 fix (off-by-one analytics drops today): shared
        # day-axis helper anchors the SQL window and bucket loop to the
        # same N day-aligned interval ending on today. Previously the
        # bucket loop computed [today-N … today-1] from window_start
        # while the SQL filter included today, silently dropping
        # today's audit events from the chart.
        axis = day_axis(window_days=window_days, now=now)
        window_start = axis.window_start

        # ── Headline counters ─────────────────────────────────────
        total_logs = (
            await self.db.execute(
                select(func.count()).select_from(AuditLog)
            )
        ).scalar() or 0

        today_start = now.replace(hour=0, minute=0, second=0, microsecond=0)
        logs_today = (
            await self.db.execute(
                select(func.count()).select_from(AuditLog)
                .where(AuditLog.created_at >= today_start)
            )
        ).scalar() or 0
        logs_window = (
            await self.db.execute(
                select(func.count()).select_from(AuditLog)
                .where(AuditLog.created_at >= window_start)
            )
        ).scalar() or 0

        unique_actors_window = (
            await self.db.execute(
                select(func.count(func.distinct(AuditLog.actor_user_id)))
                .where(
                    AuditLog.actor_user_id.isnot(None),
                    AuditLog.created_at >= window_start,
                )
            )
        ).scalar() or 0
        unique_actors_total = (
            await self.db.execute(
                select(func.count(func.distinct(AuditLog.actor_user_id)))
                .where(AuditLog.actor_user_id.isnot(None))
            )
        ).scalar() or 0

        # ── Breakdowns inside the window ──────────────────────────
        status_rows = (
            await self.db.execute(
                select(AuditLog.status, func.count())
                .where(AuditLog.created_at >= window_start)
                .group_by(AuditLog.status)
            )
        ).all()
        by_status_window = {row[0]: int(row[1]) for row in status_rows}

        actor_type_rows = (
            await self.db.execute(
                select(AuditLog.actor_type, func.count())
                .where(AuditLog.created_at >= window_start)
                .group_by(AuditLog.actor_type)
            )
        ).all()
        by_actor_type = {row[0]: int(row[1]) for row in actor_type_rows}

        # Top 20 actions in the window.
        action_rows = (
            await self.db.execute(
                select(AuditLog.action, func.count().label("c"))
                .where(AuditLog.created_at >= window_start)
                .group_by(AuditLog.action)
                .order_by(func.count().desc())
                .limit(20)
            )
        ).all()
        top_actions = [
            {"action": row.action, "count": int(row.c)} for row in action_rows
        ]

        # Top 20 resource types in the window.
        rtype_rows = (
            await self.db.execute(
                select(AuditLog.resource_type, func.count().label("c"))
                .where(
                    AuditLog.created_at >= window_start,
                    AuditLog.resource_type.isnot(None),
                )
                .group_by(AuditLog.resource_type)
                .order_by(func.count().desc())
                .limit(20)
            )
        ).all()
        top_resource_types = [
            {"resource_type": row.resource_type, "count": int(row.c)}
            for row in rtype_rows
        ]

        # ── Action-prefix bucket (helps spot which domain is hot) ─
        # Splits e.g. ``identity.person.update`` → ``identity`` so the
        # frontend can render an at-a-glance pie even when individual
        # actions are long-tail.
        prefix_rows = (
            await self.db.execute(
                select(
                    func.substring_index(AuditLog.action, ".", 1).label("p"),
                    func.count(),
                )
                .where(AuditLog.created_at >= window_start)
                .group_by("p")
                .order_by(func.count().desc())
                .limit(15)
            )
        ).all()
        by_action_prefix = [
            {"prefix": row[0] or "", "count": int(row[1])}
            for row in prefix_rows
        ]

        # ── Daily timeseries (status × day) ───────────────────────
        daily_rows = (
            await self.db.execute(
                select(
                    func.date(AuditLog.created_at).label("d"),
                    AuditLog.status,
                    func.count(),
                )
                .where(AuditLog.created_at >= window_start)
                .group_by("d", AuditLog.status)
                .order_by("d")
            )
        ).all()
        daily_map: dict[str, dict[str, int]] = {}
        for d, status_v, n in daily_rows:
            k = d.isoformat() if hasattr(d, "isoformat") else str(d)
            bucket = daily_map.setdefault(k, {"success": 0, "failure": 0})
            bucket[status_v] = int(n)
        daily_series: list[dict] = []
        for day in axis.days:
            key = day.isoformat()
            slot = daily_map.get(key, {})
            daily_series.append({
                "date": key,
                "success": slot.get("success", 0),
                "failure": slot.get("failure", 0),
                "total": slot.get("success", 0) + slot.get("failure", 0),
            })

        # ── Hourly heat (24 bins over window) ─────────────────────
        hourly_rows = (
            await self.db.execute(
                select(
                    func.hour(AuditLog.created_at).label("h"),
                    func.count(),
                )
                .where(AuditLog.created_at >= window_start)
                .group_by("h")
                .order_by("h")
            )
        ).all()
        hourly_map = {int(h): int(n) for h, n in hourly_rows}
        hourly_distribution = [
            {"hour": h, "count": hourly_map.get(h, 0)} for h in range(24)
        ]

        # ── Top human actors (window) — join identity_persons so the
        # frontend renders names not opaque UUIDs.
        actor_rows = (
            await self.db.execute(
                select(
                    AuditLog.actor_user_id,
                    func.count().label("c"),
                    func.sum(
                        case((AuditLog.status == "failure", 1), else_=0)
                    ).label("fail_count"),
                )
                .where(
                    AuditLog.created_at >= window_start,
                    AuditLog.actor_user_id.isnot(None),
                    AuditLog.actor_type == "user",
                )
                .group_by(AuditLog.actor_user_id)
                .order_by(func.count().desc())
                .limit(10)
            )
        ).all()
        actor_ids = [row[0] for row in actor_rows if row[0]]
        person_map: dict[str, tuple[str | None, str | None, str | None, str | None]] = {}
        if actor_ids:
            res = await self.db.execute(
                select(
                    IdentityPerson.id,
                    IdentityPerson.display_name,
                    IdentityPerson.work_email,
                    IdentityPerson.employee_no,
                    IdentityPerson.employment_status,
                ).where(IdentityPerson.id.in_(actor_ids))
            )
            person_map = {
                str(r.id): (r.display_name, r.work_email, r.employee_no, r.employment_status)
                for r in res
            }
        top_actors = []
        for actor_id, c, fail_c in actor_rows:
            display, email, emp_no, emp_status = person_map.get(
                str(actor_id), (None, None, None, None),
            )
            top_actors.append({
                "actor_user_id": str(actor_id) if actor_id else None,
                "actor_display_name": display,
                "actor_email": email,
                "actor_employee_no": emp_no,
                "actor_employment_status": emp_status,
                "count": int(c),
                "failure_count": int(fail_c or 0),
            })

        return {
            "window_days": window_days,
            "checked_at": now.isoformat(),
            "totals": {
                "total_logs": int(total_logs),
                "logs_today": int(logs_today),
                "logs_window": int(logs_window),
                "unique_actors_window": int(unique_actors_window),
                "unique_actors_total": int(unique_actors_total),
                "success_window": int(by_status_window.get("success", 0)),
                "failure_window": int(by_status_window.get("failure", 0)),
            },
            "by_status_window": by_status_window,
            "by_actor_type": by_actor_type,
            "top_actions": top_actions,
            "top_resource_types": top_resource_types,
            "by_action_prefix": by_action_prefix,
            "daily_series": daily_series,
            "hourly_distribution": hourly_distribution,
            "top_actors": top_actors,
        }

    async def get_distinct_actions(self) -> list[str]:
        result = await self.db.execute(
            select(AuditLog.action)
            .distinct()
            .order_by(AuditLog.action)
        )
        return [row[0] for row in result.all()]

    async def get_distinct_resource_types(self) -> list[str]:
        result = await self.db.execute(
            select(AuditLog.resource_type)
            .where(AuditLog.resource_type.isnot(None))
            .distinct()
            .order_by(AuditLog.resource_type)
        )
        return [row[0] for row in result.all()]