"""Two-phase authorization for identity-sync mass deprovisioning.

Phase 1 records the exact target/config/subject snapshot that tripped a
mass-action brake. Phase 2 lets a global admin confirm that immutable
fingerprint and returns a short-lived bearer token. The next sweep recomputes
the snapshot and atomically consumes the token before any vendor destructive
call. Drift, expiry, tampering, concurrent confirmation, and replay all fail
closed.
"""

from __future__ import annotations

import hashlib
import json
import secrets
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from typing import Any
from uuid import UUID

from sqlalchemy import desc, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.core.exceptions import AppException, NotFoundError
from app.domains.identity_sync.models import (
    IdentitySyncDestructivePlan,
    IdentitySyncTarget,
)
from app.integrations.models import Integration

DESTRUCTIVE_PLAN_TTL = timedelta(hours=24)
DESTRUCTIVE_CONFIRMATION_TTL = timedelta(minutes=10)

_ACTIVE_STATUSES = ("pending", "confirmed")


@dataclass(frozen=True, slots=True)
class DestructiveGuardResult:
    authorized: bool
    plan_id: UUID
    fingerprint: str
    reason: str


def _aware(value: datetime) -> datetime:
    return value if value.tzinfo is not None else value.replace(tzinfo=UTC)


def _canonical_hash(value: Any) -> str:
    payload = json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
        ensure_ascii=False,
        default=str,
    ).encode("utf-8")
    return hashlib.sha256(payload).hexdigest()


def target_destructive_snapshot(target: IdentitySyncTarget) -> dict[str, Any]:
    """Only fields able to change destructive scope/action/vendor identity."""

    return {
        "target_id": str(target.id),
        "kind": target.kind,
        "integration_id": str(target.integration_id) if target.integration_id else None,
        "enabled": bool(target.enabled),
        "on_offboard": target.on_offboard,
        "write_mode": target.write_mode,
        "filter_config": target.filter_config or {},
        "mapping_config": target.mapping_config or {},
        "runner_options": target.runner_options or {},
    }


def integration_destructive_snapshot(
    integration: Integration | None,
    *,
    integration_id: UUID,
) -> dict[str, Any]:
    """Non-secret downstream identity/version bound to an approval.

    ``integration_id`` alone is not a tenant boundary: administrators may
    edit the referenced row in place and point the same target at another
    endpoint or credential set.  Hash public configuration and encrypted
    credential material so either change invalidates an already-reviewed
    destructive plan without persisting credentials in the plan itself.
    """
    if integration is None:
        return {"integration_id": str(integration_id), "missing": True}
    credential_revision = _canonical_hash(
        {
            "ciphertext": integration.secrets_encrypted,
            "nonce": integration.secrets_nonce,
        }
    )
    return {
        "integration_id": str(integration.id),
        "kind": integration.kind,
        "provider": integration.provider,
        "slug": integration.slug,
        "enabled": bool(integration.enabled),
        "updated_at": integration.updated_at.isoformat(),
        "config_public_hash": _canonical_hash(integration.config_public or {}),
        "capabilities_hash": _canonical_hash(integration.capabilities or []),
        "credential_revision_hash": credential_revision,
    }


def user_mass_action_brake_required(*, out_count: int, in_scope_count: int) -> bool:
    threshold = max(25, int((in_scope_count + out_count) * 0.10))
    full_drop = in_scope_count == 0 and out_count >= 5
    return out_count > threshold or full_drop


def department_mass_action_brake_required(out_count: int) -> bool:
    return out_count > 10


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

    @staticmethod
    def hash_token(token: str) -> str:
        return hashlib.sha256(token.encode("utf-8")).hexdigest()

    async def _target_snapshot(
        self,
        target: IdentitySyncTarget,
    ) -> dict[str, Any]:
        # A sync run commonly loads both rows while constructing its runner.
        # Force a database refresh here so an administrator's concurrent
        # config/credential edit cannot be hidden by SQLAlchemy's identity
        # map. Lock both rows through the authorization commit; the runner
        # then executes with exactly the tenant snapshot that was consumed.
        fresh_target = (
            (
                await self.db.execute(
                    select(IdentitySyncTarget)
                    .where(IdentitySyncTarget.id == target.id)
                    .execution_options(populate_existing=True)
                    .with_for_update()
                )
            )
            .scalars()
            .first()
        )
        if fresh_target is None:
            return {"target_id": str(target.id), "missing": True}

        snapshot = target_destructive_snapshot(fresh_target)
        if fresh_target.integration_id is not None:
            integration = (
                (
                    await self.db.execute(
                        select(Integration)
                        .where(Integration.id == fresh_target.integration_id)
                        .execution_options(populate_existing=True)
                        .with_for_update()
                    )
                )
                .scalars()
                .first()
            )
            snapshot["integration"] = integration_destructive_snapshot(
                integration,
                integration_id=fresh_target.integration_id,
            )
        return snapshot

    async def _snapshot(
        self,
        *,
        target: IdentitySyncTarget,
        scope: str,
        action: str,
        items: list[dict[str, Any]],
    ) -> tuple[str, str, list[dict[str, Any]]]:
        target_snapshot = await self._target_snapshot(target)
        target_hash = _canonical_hash(target_snapshot)
        normalized_items = sorted(
            items,
            key=lambda item: json.dumps(
                item,
                sort_keys=True,
                separators=(",", ":"),
                ensure_ascii=False,
                default=str,
            ),
        )
        fingerprint = _canonical_hash(
            {
                "target": target_snapshot,
                "scope": scope,
                "action": action,
                "items": normalized_items,
            }
        )
        return fingerprint, target_hash, normalized_items

    async def _audit(
        self,
        *,
        trace_id: str,
        actor_user_id: UUID | None,
        action: str,
        plan: IdentitySyncDestructivePlan,
        status: str,
        reason: str | None = None,
    ) -> None:
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user" if actor_user_id else "system",
            actor_user_id=actor_user_id,
            action=action,
            resource_type="identity_sync_destructive_plan",
            resource_id=str(plan.id),
            status=status,
            detail={
                "target_id": str(plan.target_id),
                "scope": plan.scope,
                "destructive_action": plan.action,
                "fingerprint": plan.fingerprint,
                "item_count": plan.item_count,
                "reason": reason,
            },
        )

    async def _new_or_active_plan(
        self,
        *,
        target: IdentitySyncTarget,
        scope: str,
        action: str,
        fingerprint: str,
        target_hash: str,
        items: list[dict[str, Any]],
        run_id: UUID | None,
        now: datetime,
    ) -> IdentitySyncDestructivePlan:
        active_rows = list(
            (
                await self.db.execute(
                    select(IdentitySyncDestructivePlan)
                    .where(
                        IdentitySyncDestructivePlan.target_id == target.id,
                        IdentitySyncDestructivePlan.scope == scope,
                        IdentitySyncDestructivePlan.status.in_(_ACTIVE_STATUSES),
                    )
                    .order_by(desc(IdentitySyncDestructivePlan.created_at))
                    .with_for_update()
                )
            )
            .scalars()
            .all()
        )
        reusable: IdentitySyncDestructivePlan | None = None
        for row in active_rows:
            if _aware(row.plan_expires_at) <= now:
                row.status = "expired"
                continue
            if row.fingerprint == fingerprint and reusable is None:
                reusable = row
                continue
            row.status = "superseded"
        if reusable is not None:
            # An expired confirmation returns the plan to pending so the
            # administrator can explicitly confirm the still-current snapshot
            # again. Never silently extend an old bearer token.
            if (
                reusable.status == "confirmed"
                and reusable.confirmation_expires_at is not None
                and _aware(reusable.confirmation_expires_at) <= now
            ):
                reusable.status = "pending"
                reusable.confirmation_token_hash = None
                reusable.confirmed_by = None
                reusable.confirmed_at = None
                reusable.confirmation_expires_at = None
            return reusable

        plan = IdentitySyncDestructivePlan(
            target_id=target.id,
            scope=scope,
            action=action,
            fingerprint=fingerprint,
            target_snapshot_hash=target_hash,
            item_count=len(items),
            item_snapshot=items,
            status="pending",
            plan_expires_at=now + DESTRUCTIVE_PLAN_TTL,
            created_run_id=run_id,
        )
        self.db.add(plan)
        await self.db.flush()
        return plan

    async def guard_or_record(
        self,
        *,
        target: IdentitySyncTarget,
        scope: str,
        action: str,
        items: list[dict[str, Any]],
        run_id: UUID | None,
        confirmation_token_hashes: list[str],
        actor_user_id: UUID | None,
        trace_id: str,
    ) -> DestructiveGuardResult:
        """Consume an exact matching confirmation or persist a blocked plan.

        This method commits before returning. On authorization that commit is
        the one-time consumption fence and happens before the caller invokes a
        destructive vendor method.
        """

        now = datetime.now(UTC)
        fingerprint, target_hash, normalized_items = await self._snapshot(
            target=target,
            scope=scope,
            action=action,
            items=items,
        )
        matched: IdentitySyncDestructivePlan | None = None
        if confirmation_token_hashes:
            matched = (
                (
                    await self.db.execute(
                        select(IdentitySyncDestructivePlan)
                        .where(
                            IdentitySyncDestructivePlan.target_id == target.id,
                            IdentitySyncDestructivePlan.scope == scope,
                            IdentitySyncDestructivePlan.confirmation_token_hash.in_(
                                confirmation_token_hashes
                            ),
                        )
                        .order_by(desc(IdentitySyncDestructivePlan.created_at))
                        .with_for_update()
                    )
                )
                .scalars()
                .first()
            )

        rejection_reason: str | None = None
        if matched is not None:
            if matched.status == "consumed":
                rejection_reason = "token_replayed"
            elif matched.status != "confirmed":
                rejection_reason = "invalid_token"
            elif (
                matched.confirmation_expires_at is None
                or _aware(matched.confirmation_expires_at) <= now
            ):
                matched.status = "expired"
                rejection_reason = "token_expired"
            elif _aware(matched.plan_expires_at) <= now:
                matched.status = "expired"
                rejection_reason = "plan_expired"
            elif matched.fingerprint != fingerprint or matched.target_snapshot_hash != target_hash:
                matched.status = "superseded"
                rejection_reason = "snapshot_drifted"
            else:
                matched.status = "consumed"
                matched.consumed_run_id = run_id
                matched.consumed_by = actor_user_id
                matched.consumed_at = now
                await self._audit(
                    trace_id=trace_id,
                    actor_user_id=actor_user_id,
                    action="identity_sync.destructive_plan.consume",
                    plan=matched,
                    status="success",
                )
                await self.db.commit()
                return DestructiveGuardResult(
                    authorized=True,
                    plan_id=matched.id,
                    fingerprint=fingerprint,
                    reason="authorized",
                )
        elif confirmation_token_hashes:
            rejection_reason = "invalid_token"

        plan = await self._new_or_active_plan(
            target=target,
            scope=scope,
            action=action,
            fingerprint=fingerprint,
            target_hash=target_hash,
            items=normalized_items,
            run_id=run_id,
            now=now,
        )
        reason = rejection_reason or "confirmation_required"
        await self._audit(
            trace_id=trace_id,
            actor_user_id=actor_user_id,
            action="identity_sync.destructive_plan.block",
            plan=plan,
            status="failure",
            reason=reason,
        )
        await self.db.commit()
        return DestructiveGuardResult(
            authorized=False,
            plan_id=plan.id,
            fingerprint=fingerprint,
            reason=reason,
        )

    async def confirm(
        self,
        *,
        plan_id: UUID,
        expected_fingerprint: str,
        actor_user_id: UUID,
        trace_id: str,
        reissue: bool = False,
    ) -> str:
        now = datetime.now(UTC)
        plan = (
            (
                await self.db.execute(
                    select(IdentitySyncDestructivePlan)
                    .where(IdentitySyncDestructivePlan.id == plan_id)
                    .with_for_update()
                )
            )
            .scalars()
            .first()
        )
        if plan is None:
            raise NotFoundError("IdentitySyncDestructivePlan", str(plan_id))
        target = await self.db.get(IdentitySyncTarget, plan.target_id)
        if target is None or target.deleted_at is not None:
            plan.status = "superseded"
            await self.db.commit()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_DRIFTED",
                message="Sync target no longer exists; generate a new destructive plan",
                status_code=409,
            )
        if expected_fingerprint != plan.fingerprint:
            await self._audit(
                trace_id=trace_id,
                actor_user_id=actor_user_id,
                action="identity_sync.destructive_plan.confirm",
                plan=plan,
                status="failure",
                reason="fingerprint_mismatch",
            )
            await self.db.commit()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_FINGERPRINT_MISMATCH",
                message="Plan fingerprint does not match the reviewed snapshot",
                status_code=409,
            )
        if _aware(plan.plan_expires_at) <= now:
            plan.status = "expired"
            await self.db.commit()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_EXPIRED",
                message="Destructive plan expired; run a new preview",
                status_code=409,
            )
        current_target_hash = _canonical_hash(
            await self._target_snapshot(target),
        )
        if current_target_hash != plan.target_snapshot_hash:
            plan.status = "superseded"
            await self._audit(
                trace_id=trace_id,
                actor_user_id=actor_user_id,
                action="identity_sync.destructive_plan.confirm",
                plan=plan,
                status="failure",
                reason="target_config_drifted",
            )
            await self.db.commit()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_DRIFTED",
                message="Target configuration changed; generate and review a new plan",
                status_code=409,
            )
        if plan.status == "confirmed" and not reissue:
            await self.db.rollback()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_ALREADY_CONFIRMED",
                message="Plan already confirmed; explicitly reissue if the token was lost",
                status_code=409,
            )
        if reissue and plan.status != "confirmed":
            await self.db.rollback()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_NOT_REISSUABLE",
                message="Only a confirmed plan with a lost token can be reissued",
                status_code=409,
            )
        if plan.status not in _ACTIVE_STATUSES:
            await self.db.rollback()
            raise AppException(
                code="IDENTITY_SYNC_DESTRUCTIVE_PLAN_NOT_CONFIRMABLE",
                message=f"Plan status {plan.status!r} cannot be confirmed",
                status_code=409,
            )

        token = f"idsync_{secrets.token_urlsafe(32)}"
        plan.status = "confirmed"
        plan.confirmation_token_hash = self.hash_token(token)
        plan.confirmed_by = actor_user_id
        plan.confirmed_at = now
        plan.confirmation_expires_at = now + DESTRUCTIVE_CONFIRMATION_TTL
        await self._audit(
            trace_id=trace_id,
            actor_user_id=actor_user_id,
            action=(
                "identity_sync.destructive_plan.reissue"
                if reissue
                else "identity_sync.destructive_plan.confirm"
            ),
            plan=plan,
            status="success",
        )
        await self.db.commit()
        return token

    async def list_active(self, target_id: UUID) -> list[IdentitySyncDestructivePlan]:
        rows = list(
            (
                await self.db.execute(
                    select(IdentitySyncDestructivePlan)
                    .where(
                        IdentitySyncDestructivePlan.target_id == target_id,
                        IdentitySyncDestructivePlan.status.in_(_ACTIVE_STATUSES),
                    )
                    .order_by(desc(IdentitySyncDestructivePlan.created_at))
                )
            )
            .scalars()
            .all()
        )
        now = datetime.now(UTC)
        live: list[IdentitySyncDestructivePlan] = []
        changed = False
        for row in rows:
            if _aware(row.plan_expires_at) <= now:
                row.status = "expired"
                changed = True
            else:
                live.append(row)
        if changed:
            await self.db.commit()
        return live