"""入职须知 (onboarding acknowledgement) service.

Two surfaces:

* Admin CRUD on the checklist items (``onboarding_ack_items``) — what a
  new hire must read & confirm before Day-1 readiness is complete.
* Employee confirm (``acknowledge``) — idempotent per (item, person);
  immediately rescans the caller's open onboarding process so the
  ``training_ack`` blocker resolves on click instead of waiting for the
  5-min reconcile cron.

Deleting an item that already has acknowledgements is refused (409) —
those rows are compliance evidence; deactivate instead.
"""
from __future__ import annotations

from typing import Any
from uuid import UUID

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

from app.audit.service import AuditService
from app.core.exceptions import AppException
from app.domains.onboarding.models import (
    ACK_SOURCE_WEB,
    PROC_OPEN_STATUSES,
    OnboardingAckItem,
    OnboardingAcknowledgement,
    OnboardingProcess,
)

_VALID_ACK_SOURCES = ("web", "mobile", "mcp")


def _validate_doc_url(value: str | None) -> str | None:
    """Defence-in-depth: the link renders as a clickable CTA for every
    new hire — only http(s) schemes are storable (javascript:/data:
    etc. refused even though writers are global_admin-gated)."""
    if value is None:
        return None
    if not value.lower().startswith(("https://", "http://")):
        raise AppException(
            code="ACK_ITEM_DOC_URL_SCHEME",
            message="doc_url must start with https:// or http://",
            status_code=400,
        )
    return value


class OnboardingAckService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.audit = AuditService(db)

    # ───────────────────────── Admin CRUD ─────────────────────────

    async def list_items(
        self, *, include_inactive: bool = False,
    ) -> list[OnboardingAckItem]:
        query = select(OnboardingAckItem).order_by(
            OnboardingAckItem.sort_order, OnboardingAckItem.created_at,
        )
        if not include_inactive:
            query = query.where(OnboardingAckItem.is_active.is_(True))
        return list((await self.db.execute(query)).scalars().all())

    async def create_item(
        self,
        *,
        title: str,
        description: str | None,
        doc_url: str | None,
        required: bool,
        sort_order: int,
        trace_id: str,
        actor_user_id: UUID,
    ) -> OnboardingAckItem:
        clean_doc_url = _validate_doc_url((doc_url or "").strip() or None)
        item = OnboardingAckItem(
            title=title.strip(),
            description=(description or "").strip() or None,
            doc_url=clean_doc_url,
            required=required,
            sort_order=sort_order,
            created_by_person_id=actor_user_id,
        )
        self.db.add(item)
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="onboarding.ack_item_created",
            status="success",
            resource_type="onboarding_ack_item",
            resource_id=str(item.id),
            detail={"title": item.title, "required": required},
        )
        return item

    async def update_item(
        self,
        item_id: UUID,
        *,
        updates: dict[str, Any],
        trace_id: str,
        actor_user_id: UUID,
    ) -> OnboardingAckItem:
        item = await self._get_item_or_raise(item_id)
        changes: dict[str, Any] = {}
        for field in (
            "title", "description", "doc_url",
            "required", "is_active", "sort_order",
        ):
            if field not in updates:
                continue
            value = updates[field]
            if field in ("title", "description", "doc_url") and value is not None:
                value = str(value).strip() or None
            if field == "title" and not value:
                raise AppException(
                    code="ACK_ITEM_TITLE_REQUIRED",
                    message="title cannot be empty",
                    status_code=400,
                )
            # NOT NULL columns: an explicit JSON null would survive
            # Pydantic's exclude_unset and 500 at flush — refuse with a
            # typed 400 instead (review-r1).
            if field in ("required", "is_active", "sort_order") and value is None:
                raise AppException(
                    code="ACK_ITEM_FIELD_NOT_NULLABLE",
                    message=f"{field} cannot be null",
                    status_code=400,
                )
            if field == "doc_url":
                value = _validate_doc_url(value)
            if getattr(item, field) != value:
                changes[field] = {
                    "from": getattr(item, field), "to": value,
                }
                setattr(item, field, value)
        if changes:
            await self.db.flush()
            await self.audit.log(
                trace_id=trace_id,
                actor_type="user",
                actor_user_id=actor_user_id,
                action="onboarding.ack_item_updated",
                status="success",
                resource_type="onboarding_ack_item",
                resource_id=str(item.id),
                detail=changes,
            )
        return item

    async def delete_item(
        self,
        item_id: UUID,
        *,
        trace_id: str,
        actor_user_id: UUID,
    ) -> None:
        item = await self._get_item_or_raise(item_id)
        ack_count = int(
            (
                await self.db.execute(
                    select(func.count())
                    .select_from(OnboardingAcknowledgement)
                    .where(OnboardingAcknowledgement.item_id == item_id)
                )
            ).scalar_one()
        )
        if ack_count > 0:
            # Acks are compliance evidence ("the hire confirmed reading
            # the security policy") — a hard delete CASCADEs them away.
            raise AppException(
                code="ACK_ITEM_HAS_ACKNOWLEDGEMENTS",
                message=(
                    f"{ack_count} acknowledgement(s) reference this item; "
                    "deactivate it instead of deleting"
                ),
                status_code=409,
            )
        title = item.title
        await self.db.delete(item)
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor_user_id,
            action="onboarding.ack_item_deleted",
            status="success",
            resource_type="onboarding_ack_item",
            resource_id=str(item_id),
            detail={"title": title},
        )

    # ───────────────────────── Employee surface ─────────────────────────

    async def my_ack_status(self, person_id: UUID) -> list[dict[str, Any]]:
        """Active checklist items + whether the caller confirmed each."""
        items = await self.list_items(include_inactive=False)
        if not items:
            return []
        acked_at_by_item: dict[UUID, Any] = {
            row.item_id: row.acknowledged_at
            for row in (
                await self.db.execute(
                    select(OnboardingAcknowledgement).where(
                        OnboardingAcknowledgement.person_id == person_id,
                        OnboardingAcknowledgement.item_id.in_(
                            [i.id for i in items]
                        ),
                    )
                )
            ).scalars()
        }
        return [
            {
                "id": str(i.id),
                "title": i.title,
                "description": i.description,
                "doc_url": i.doc_url,
                "required": bool(i.required),
                "acknowledged": i.id in acked_at_by_item,
                "acknowledged_at": (
                    acked_at_by_item[i.id].isoformat()
                    if i.id in acked_at_by_item
                    and acked_at_by_item[i.id] is not None
                    else None
                ),
            }
            for i in items
        ]

    async def acknowledge(
        self,
        person_id: UUID,
        item_id: UUID,
        *,
        source: str = ACK_SOURCE_WEB,
        trace_id: str,
    ) -> OnboardingAcknowledgement:
        """Idempotent confirm. Re-acking returns the existing row.

        Rescans the caller's open onboarding process (if any) so the
        ``training_ack`` blocker resolves immediately.
        """
        item = await self._get_item_or_raise(item_id)
        if not item.is_active:
            raise AppException(
                code="ACK_ITEM_INACTIVE",
                message="this item is no longer active",
                status_code=409,
            )
        if source not in _VALID_ACK_SOURCES:
            # The source column is compliance evidence — refuse rather
            # than silently coerce (HTTP regex + MCP constant make this
            # unreachable today; this backstops future callers).
            raise AppException(
                code="ACK_SOURCE_INVALID",
                message=f"source must be one of {_VALID_ACK_SOURCES}",
                status_code=400,
            )

        existing = (
            await self.db.execute(
                select(OnboardingAcknowledgement).where(
                    OnboardingAcknowledgement.item_id == item_id,
                    OnboardingAcknowledgement.person_id == person_id,
                )
            )
        ).scalar_one_or_none()
        if existing is not None:
            return existing

        open_proc = (
            await self.db.execute(
                select(OnboardingProcess).where(
                    OnboardingProcess.person_id == person_id,
                    OnboardingProcess.status.in_(PROC_OPEN_STATUSES),
                )
            )
        ).scalar_one_or_none()

        ack = OnboardingAcknowledgement(
            item_id=item_id,
            person_id=person_id,
            process_id=open_proc.id if open_proc is not None else None,
            source=source,
        )
        try:
            async with self.db.begin_nested():
                self.db.add(ack)
                await self.db.flush()
        except IntegrityError:
            # Double-click / concurrent MCP+web race — the UNIQUE
            # constraint won; reload and return the winner.
            existing = (
                await self.db.execute(
                    select(OnboardingAcknowledgement).where(
                        OnboardingAcknowledgement.item_id == item_id,
                        OnboardingAcknowledgement.person_id == person_id,
                    )
                )
            ).scalar_one_or_none()
            if existing is None:
                raise
            return existing

        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=person_id,
            action="onboarding.item_acknowledged",
            status="success",
            resource_type="onboarding_ack_item",
            resource_id=str(item_id),
            detail={"title": item.title, "source": source},
        )

        if open_proc is not None:
            # Resolve the training_ack blocker on click rather than on
            # the next 5-min reconcile tick.
            from app.domains.onboarding.service import OnboardingService

            await OnboardingService(self.db).reconcile_blockers(open_proc.id)
        return ack

    # ───────────────────────── Helpers ─────────────────────────

    async def _get_item_or_raise(self, item_id: UUID) -> OnboardingAckItem:
        item = await self.db.get(OnboardingAckItem, item_id)
        if item is None:
            raise AppException(
                code="ACK_ITEM_NOT_FOUND",
                message=f"ack item {item_id} not found",
                status_code=404,
            )
        return item


def serialize_ack_item(item: OnboardingAckItem) -> dict[str, Any]:
    return {
        "id": str(item.id),
        "title": item.title,
        "description": item.description,
        "doc_url": item.doc_url,
        "required": bool(item.required),
        "is_active": bool(item.is_active),
        "sort_order": int(item.sort_order),
        "created_at": (
            item.created_at.isoformat() if item.created_at else None
        ),
        "updated_at": (
            item.updated_at.isoformat() if item.updated_at else None
        ),
    }