"""Deterministic generated-artifact checks for the admitted auth contract.

This is deliberately a source conformance gate, not an auth runtime.  Runtime
and Cloud tests prove execution; this module prevents generated source from
quietly adding an auth method or bypassing the canonical client surface.
"""

from __future__ import annotations

import json
import os
import re
from dataclasses import dataclass
from pathlib import Path
from typing import Any


_SKIP_DIRS = {".git", ".p2hs", "node_modules", ".nuxt", ".output", "dist", "build"}
_SKIP_FILES = {"p2hs-codegen-action-runner.mjs"}
_SOURCE_SUFFIXES = {".html", ".htm", ".js", ".mjs", ".cjs", ".ts", ".tsx", ".jsx", ".vue"}
# Seeded by Core's staff shell installer. It uses the canonical magic-link
# helper by design and is not provider-authored customer UI.
_CORE_STAFF_SHELL_PATHS = {"src/components/P2hsStaffDashboard.jsx"}


@dataclass(frozen=True)
class AuthConformanceFinding:
    rule_id: str
    severity: str
    artifact: str
    line: int
    expected: str
    observed: str


def _contract(plan: dict[str, Any]) -> dict[str, Any] | None:
    runtime = plan.get("runtime") or {}
    if not runtime.get("authentication_required"):
        return None
    return runtime.get("auth_contract") or {}


def _methods(contract: dict[str, Any]) -> set[str]:
    return {
        str(method)
        for audience in contract.get("audiences") or []
        for method in audience.get("methods") or []
    }


def _audience_flags(contract: dict[str, Any]) -> tuple[bool, bool, bool]:
    audiences = contract.get("audiences") or []
    customer = next((item for item in audiences if item.get("audience") == "customer"), None)
    customer_signup = bool(customer and customer.get("signup_allowed"))
    password_reset = any(bool(item.get("password_reset_allowed")) for item in audiences)
    return bool(customer), customer_signup, password_reset


def _audiences(contract: dict[str, Any]) -> dict[str, dict[str, Any]]:
    """Return the admitted auth operations keyed by generated-app audience."""
    return {
        str(item.get("audience", "")).strip().lower(): item
        for item in contract.get("audiences") or []
        if isinstance(item, dict) and item.get("audience")
    }


def _is_staff_surface(rel: str, staff_route: str) -> bool:
    """Identify generated staff UI without treating admin API handlers as UI."""
    normalized = "/" + rel.replace("\\", "/").lstrip("/").lower()
    if normalized.startswith(("/server/api/", "/functions/api/", "/api/")):
        return False

    route = "/" + str(staff_route or "/admin/login").lstrip("/").lower()
    route_parts = [part for part in route.split("/") if part]
    if not route_parts:
        return False
    # The login route is the most precise contract marker.  Its parent route
    # covers the sibling forgot/reset pages that must disappear together with
    # a passwordless staff login.
    parent = route_parts[:-1] or route_parts
    rel_parts = [part for part in normalized.split("/") if part]
    if any(
        rel_parts[i : i + len(parent)] == parent
        for i in range(len(rel_parts) - len(parent) + 1)
    ):
        return True
    return any(token in normalized for token in ("/admin.html", "/admin/", "/be-admin/"))


def legacy_migration_generation_plan(auth_target: dict[str, Any]) -> dict[str, Any]:
    """Adapt a certified legacy target into the existing AUTH-GEN shape.

    This is an ephemeral conformance input; it does not fabricate historical
    ``generation-plan.json`` provenance.
    """
    audiences = []
    for name in ("staff", "customer"):
        audience = auth_target.get(name) or {}
        methods = list(audience.get("methods") or [])
        if "passwordless" in methods and "magic_link" not in methods:
            methods.append("magic_link")
        audiences.append(
            {
                "audience": name,
                "methods": methods,
                "signup_allowed": bool(audience.get("signup")),
                "password_reset_allowed": bool(audience.get("password_reset")),
            }
        )
    return {
        "runtime": {
            "authentication_required": True,
            "auth_contract": {"audiences": audiences},
        }
    }


def check_legacy_migration_auth_conformance(
    repo_path: str | Path,
    auth_target: dict[str, Any],
    managed_paths: set[str] | None = None,
) -> list[AuthConformanceFinding]:
    """Run the existing source scanner for a certified legacy target."""
    return check_generated_auth_conformance(
        repo_path,
        legacy_migration_generation_plan(auth_target),
        managed_paths=managed_paths,
    )


def _line_number(source: str, match: re.Match[str]) -> int:
    return source.count("\n", 0, match.start()) + 1


def _finding(rule: str, path: Path, source: str, match: re.Match[str], expected: str, observed: str) -> AuthConformanceFinding:
    return AuthConformanceFinding(
        rule_id=rule,
        severity="blocking",
        artifact=path.as_posix(),
        line=_line_number(source, match),
        expected=expected,
        observed=observed,
    )


def check_generated_auth_conformance(
    repo_path: str | Path,
    generation_plan: dict[str, Any],
    managed_paths: set[str] | None = None,
) -> list[AuthConformanceFinding]:
    """Return blocking findings; callers decide whether to raise or report."""

    root = Path(repo_path)
    contract = _contract(generation_plan)
    if not contract:
        return []
    methods = _methods(contract)
    has_password = "password" in methods
    has_magic = bool({"magic_link", "passwordless"} & methods)
    has_customer, customer_signup, password_reset = _audience_flags(contract)
    audiences = _audiences(contract)
    staff = audiences.get("staff") or {}
    staff_methods = {
        str(method).strip().lower() for method in staff.get("methods") or []
    }
    staff_has_password = "password" in staff_methods
    staff_has_reset = bool(staff.get("password_reset_allowed"))
    staff_route = str(staff.get("login_route") or "/admin/login")
    findings: list[AuthConformanceFinding] = []
    locked_platform_rels = _locked_platform_rels(root)

    patterns: list[tuple[str, re.Pattern[str], bool, str, str]] = [
        (
            "AUTH-GEN-001",
            re.compile(r"<input[^>]+type\s*=\s*['\"]password|password(?:Input|Field|Form)|type\s*=\s*['\"]password", re.I),
            has_password,
            "password admitted by GenerationPlanV1.auth",
            "password UI or password field",
        ),
        (
            "AUTH-GEN-002",
            re.compile(r"forgot-password|reset-password|requestPasswordReset|confirmPasswordReset|password_reset", re.I),
            password_reset,
            "password reset admitted by GenerationPlanV1.auth",
            "password reset surface",
        ),
        (
            "AUTH-GEN-003",
            re.compile(r"requestMagicLink|magic-link/request|magic_link", re.I),
            has_magic,
            "magic_link admitted by GenerationPlanV1.auth",
            "magic-link surface",
        ),
        (
            "AUTH-GEN-004",
            re.compile(r"(?:signUp|/api/auth/signup|customerSignup|customer[_-]?signup)", re.I),
            customer_signup,
            "customer signup admitted by GenerationPlanV1.auth",
            "customer signup surface",
        ),
        (
            "AUTH-GEN-005",
            re.compile(
                r"""(?:from\s+|import\s+[^;\n]+\s+from\s+|require\s*\(\s*)[\"']"""
                r"""(?:@?supabase|@?firebase|@?auth0|@?clerk)(?:/[^\"']*)?[\"']"""
                r"""|\b(?:supabase|firebase|auth0|clerk)\s*\.\s*"""
                r"""(?:auth|signIn|signUp|currentUser|user)\b""",
                re.I,
            ),
            False,
            "provider-neutral canonical auth contract",
            "provider-based auth inference",
        ),
        (
            "AUTH-GEN-006",
            re.compile(r"(?:role\s*[:=]\s*['\"](?:admin|super_admin)|set(?:Role|UserRole)|assign(?:Role|Admin))", re.I),
            False,
            "roles projected by Core membership authority",
            "client-side role assignment or escalation",
        ),
    ]

    for directory, dirs, files in os.walk(root, followlinks=False):
        dirs[:] = [item for item in dirs if item not in _SKIP_DIRS]
        for name in files:
            path = Path(directory) / name
            if (
                path.suffix.lower() not in _SOURCE_SUFFIXES
                or ".test." in name
                or ".spec." in name
                or name in _SKIP_FILES
            ):
                continue
            rel = path.relative_to(root).as_posix()
            # Installed platform adapters are Core-owned artifacts.  Their
            # canonical auth calls are intentionally outside provider-source
            # conformance scanning; backend_guard validates their digests.
            if rel in _CORE_STAFF_SHELL_PATHS or rel in locked_platform_rels or rel in (managed_paths or set()) or any(
                rel.startswith(prefix.rstrip("/") + "/") for prefix in (managed_paths or set()) if prefix.endswith("/")
            ):
                continue
            try:
                source = path.read_text(encoding="utf-8-sig")
            except (OSError, UnicodeDecodeError):
                continue
            for rule, pattern, allowed, expected, observed in patterns:
                if allowed:
                    continue
                match = pattern.search(source)
                if match:
                    findings.append(_finding(rule, Path(rel), source, match, expected, observed))
            # Auth methods are audience-scoped. A customer password method
            # must not make a stale staff/admin password shell conform when
            # staff is magic-link-only. Keep this focused on generated UI;
            # admin user-management API handlers may submit passwords to the
            # platform's canonical operation.
            if _is_staff_surface(rel, staff_route):
                if not staff_has_password:
                    match = re.search(
                        r"<input[^>]+type\s*=\s*['\"]password|"
                        r"password(?:Input|Field|Form)|type\s*=\s*['\"]password",
                        source,
                        re.I,
                    )
                    if match:
                        findings.append(_finding(
                            "AUTH-GEN-008", Path(rel), source, match,
                            "staff password method admitted by GenerationPlanV1.auth",
                            "password UI on a staff/admin surface",
                        ))
                if not staff_has_reset:
                    match = re.search(
                        r"forgot-password|reset-password|requestPasswordReset|"
                        r"confirmPasswordReset|password_reset",
                        source,
                        re.I,
                    )
                    if match:
                        findings.append(_finding(
                            "AUTH-GEN-009", Path(rel), source, match,
                            "staff password reset admitted by GenerationPlanV1.auth",
                            "password reset UI on a staff/admin surface",
                        ))
            if re.search(r"/v1/projects/[^\s'\"`]+/auth/|/api/auth/", source, re.I):
                match = re.search(r"/v1/projects/[^\s'\"`]+/auth/|/api/auth/", source, re.I)
                assert match is not None
                findings.append(_finding(
                    "AUTH-GEN-007", Path(rel), source, match,
                    "canonical SDK/client auth operation",
                    "direct auth endpoint usage in generated source",
                ))
    return findings


def _locked_platform_rels(root: Path) -> set[str]:
    path = root / "p2hs" / "platform-ports.lock.json"
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
        return set()
    files = data.get("files") if isinstance(data, dict) else None
    return {str(rel) for rel in files} if isinstance(files, dict) else set()


def canonical_generation_plan_required(repo_path: str | Path) -> bool:
    """Whether this checkout is a canonical generated artifact.

    The authoritative Core composition projection is the stable marker for a
    new generated project.  Older imported projects may have no such marker;
    those remain classifiable by the Phase 5 reconciliation flow.
    """

    root = Path(repo_path)
    path = root / "p2hs" / "app-blocks-plan.json"
    try:
        data = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
        return False
    return isinstance(data, dict) and data.get("adoption") == "authoritative"


def assert_generated_auth_conformance(repo_path: str | Path, generation_plan: dict[str, Any]) -> None:
    findings = check_generated_auth_conformance(repo_path, generation_plan)
    if findings:
        details = "\n".join(
            f"{item.rule_id} {item.artifact}:{item.line}: {item.observed}; expected {item.expected}"
            for item in findings[:50]
        )
        raise ValueError("Generated auth conformance failed:\n" + details)


def load_generation_plan_artifact(repo_path: str | Path) -> dict[str, Any] | None:
    path = Path(repo_path) / "p2hs" / "generation-plan.json"
    try:
        value = json.loads(path.read_text(encoding="utf-8-sig"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError):
        return None
    return value if isinstance(value, dict) else None