"""Jamf Pro MDM driver (Apple device management).

One integration row = one Jamf Pro cloud tenant (``https://<org>.jamfcloud
.com``), authenticated via the OAuth2 **client-credentials** flow — a Jamf
"API Role + API Client" pair, no user login. Credentials:

  * ``base_url``      — tenant base URL, e.g. ``https://pingcap.jamfcloud.com``
  * ``client_id``     — API Client's client id (GUID)
  * ``client_secret`` — API Client's generated secret (the only secret)

An MDM "seat" == one managed device. The **read** surface (``list_devices``
/ ``get_seat_usage``) powers seat reconciliation and asset matching; the
**action** surface (``lock_device`` / ``wipe_device`` /
``clear_device_assignment``) powers the leaver offboarding flow. The API
Role must grant the matching privileges — read needs *Read Computers /
Read Mobile Devices / Read User*; actions need *Send … Remote Lock/Wipe
Command* and *Update Computers / Update Mobile Devices*.

Jamf Pro API reference
    token:    POST {base}/api/oauth/token  (grant_type=client_credentials)
    macs:     GET  {base}/api/v1/computers-inventory
    mobile:   GET  {base}/api/v2/mobile-devices
    commands: POST {base}/api/v2/mdm/commands  (DEVICE_LOCK / ERASE_DEVICE)
"""
from __future__ import annotations

import asyncio
import logging
import time
from datetime import datetime
from typing import Any

import httpx

from app.integrations.driver_base import (
    MdmActionResult,
    MdmDevice,
    MdmDriver,
    MdmSeatUsage,
)
from app.integrations.registry import register_driver
from app.integrations.schemas import (
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)

_HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
_TRANSIENT_STATUSES: frozenset[int] = frozenset({429, 500, 502, 503, 504})
_RETRY_BACKOFF_SECONDS: tuple[float, ...] = (2.0, 6.0, 18.0)
_TOKEN_EXPIRY_SKEW = 60.0

# Inventory pagination. Jamf caps page-size at 2000; 200 keeps payloads
# small while bounding round-trips. The page guard is a runaway backstop
# (200 pages × 200 = 40k devices) far above a ~500-employee fleet.
_PAGE_SIZE = 200
_MAX_PAGES = 200

# Computer inventory sections we need — repeated ``section=`` query params.
_COMPUTER_SECTIONS = (
    "GENERAL",
    "HARDWARE",
    "OPERATING_SYSTEM",
    "USER_AND_LOCATION",
)


class JamfError(RuntimeError):
    """A non-transient Jamf failure the caller surfaces verbatim."""

    def __init__(self, operation: str, status: int | None, detail: str) -> None:
        super().__init__(f"{operation} failed (HTTP {status}): {detail}")
        self.operation = operation
        self.status = status
        self.detail = detail


def _parse_dt(value: Any) -> datetime | None:
    """Parse a Jamf ISO-8601 timestamp (``2026-06-25T08:30:00.000Z``)."""
    if not value or not isinstance(value, str):
        return None
    try:
        return datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None


@register_driver
class JamfDriver(MdmDriver):
    provider = "jamf"
    display_name = "Jamf Pro (MDM)"
    description = (
        "Jamf Pro(client-credentials)。读取已纳管 Apple 设备清单与席位占用"
        "(Computers / Mobile Devices),离职时锁定 / 解绑 / 恢复出厂。需要在 "
        "Jamf 后台 Settings → System → API Roles and Clients 建 API Role"
        "(Read Computers / Read Mobile Devices / Read User + Send Remote "
        "Lock/Wipe Command + Update Computers/Mobile Devices)和绑定它的 API "
        "Client。"
    )

    config_schema = [
        IntegrationFieldInfo(
            key="base_url", label="Jamf Pro URL",
            is_secret=False, required=True,
            placeholder="https://pingcap.jamfcloud.com",
            hint="Jamf Pro 租户地址,不带末尾斜杠。",
        ),
        IntegrationFieldInfo(
            key="client_id", label="API Client ID",
            is_secret=False, required=True,
            placeholder="6f7d1d42-cdf2-43e6-a3bc-6aea1e1f798a",
            hint="API Roles and Clients 里 API Client 的 Client ID。",
        ),
        IntegrationFieldInfo(
            key="client_secret", label="API Client Secret",
            is_secret=True, required=True,
            placeholder="••••••••",
            hint="API Client 生成的 Client Secret(仅显示一次)。",
        ),
    ]

    def __init__(
        self,
        *,
        integration_id: str,
        config: dict[str, Any],
        enabled_capabilities: list[str] | None = None,
    ) -> None:
        super().__init__(
            integration_id=integration_id,
            config=config,
            enabled_capabilities=enabled_capabilities,
        )
        self._token: str | None = None
        self._token_expiry: float = 0.0

    # ── config accessors ─────────────────────────────────────────
    def _cfg(self, key: str) -> str:
        v = self.config.get(key)
        if not v:
            raise ValueError(f"{key} is not configured")
        return str(v)

    @property
    def _base(self) -> str:
        return self._cfg("base_url").rstrip("/")

    # ── auth ─────────────────────────────────────────────────────
    async def _get_token(self) -> str:
        """Return a cached bearer token, fetching a fresh one when stale."""
        now = time.monotonic()
        if self._token and now < self._token_expiry - _TOKEN_EXPIRY_SKEW:
            return self._token
        # SSRF guard on the admin-configured base_url: this driver mints a
        # bearer token and fires destructive MDM commands (device wipe /
        # EraseDevice), so an operator-supplied host must be validated
        # against private/link-local ranges the same way every other
        # external driver does (successfactors.py). Enforced on the token
        # URL here and on every request URL in ``_request``.
        from app.integrations._url_safety import (  # noqa: PLC0415
            safe_async_http_transport,
        )
        token_url = f"{self._base}/api/oauth/token"
        transport = await safe_async_http_transport(
            token_url, allowed_suffixes=None,
        )
        data = {
            "grant_type": "client_credentials",
            "client_id": self._cfg("client_id"),
            "client_secret": self._cfg("client_secret"),
        }
        async with httpx.AsyncClient(
            timeout=_HTTP_TIMEOUT, trust_env=False, transport=transport,
        ) as c:
            resp = await c.post(
                token_url,
                data=data,
                headers={"Content-Type": "application/x-www-form-urlencoded"},
            )
        if resp.status_code != 200:
            raise JamfError("token", resp.status_code, resp.text[:300])
        body = resp.json()
        self._token = str(body["access_token"])
        self._token_expiry = now + float(body.get("expires_in", 1800))
        return self._token

    async def _request(
        self,
        method: str,
        path: str,
        *,
        json_body: Any | None = None,
        params: list[tuple[str, Any]] | dict[str, Any] | None = None,
        retry_on_transient: bool = True,
    ) -> httpx.Response:
        """Jamf request with bearer auth + bounded transient retry.

        Non-transient responses (incl. 4xx) are returned to the caller to
        interpret (idempotency on 404 etc.). A token-expiry 401 triggers
        one forced refresh + retry. Raises on exhausted transient retries.

        ``retry_on_transient=False`` disables the 5xx/429 backoff-retry for
        NON-IDEMPOTENT calls (Jamf MDM command POSTs have no idempotency
        key): a transient failure that arrives AFTER Jamf already queued an
        ``ERASE_DEVICE`` / ``DEVICE_LOCK`` must NOT be blindly re-POSTed, or
        the device gets two destructive commands. The 401-refresh retry is
        always kept (it re-sends nothing that Jamf acted on).
        """
        url = path if path.startswith("http") else f"{self._base}{path}"
        # SSRF guard on the resolved request URL (covers the http-path
        # branch too — a caller-supplied absolute URL must not escape to a
        # private/link-local host with our bearer token attached).
        from app.integrations._url_safety import (  # noqa: PLC0415
            PinnedAsyncHTTPTransport,
            resolve_safe_external_url_async,
        )
        target = await resolve_safe_external_url_async(
            url, allowed_suffixes=None,
        )
        attempts = len(_RETRY_BACKOFF_SECONDS) + 1
        refreshed = False
        resp: httpx.Response | None = None
        for attempt in range(attempts):
            transport = PinnedAsyncHTTPTransport(target)
            token = await self._get_token()
            headers = {
                "Authorization": f"Bearer {token}",
                "Accept": "application/json",
            }
            if json_body is not None:
                headers["Content-Type"] = "application/json"
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT, trust_env=False, transport=transport,
            ) as c:
                resp = await c.request(
                    method, url, headers=headers, json=json_body, params=params,
                )
            if resp.status_code == 401 and not refreshed:
                self._token = None
                refreshed = True
                continue
            if resp.status_code in _TRANSIENT_STATUSES and retry_on_transient and attempt < len(
                _RETRY_BACKOFF_SECONDS,
            ):
                backoff = _RETRY_BACKOFF_SECONDS[attempt]
                ra = resp.headers.get("Retry-After")
                if ra:
                    try:
                        backoff = max(backoff, float(ra))
                    except ValueError:
                        pass
                logger.warning(
                    "jamf %s %s transient HTTP %s (attempt %d/%d); "
                    "backing off %.1fs",
                    method, path, resp.status_code, attempt + 1, attempts,
                    backoff,
                )
                await asyncio.sleep(backoff)
                continue
            return resp
        assert resp is not None  # noqa: S101 — loop runs ≥1 time
        return resp

    # ── health probe ─────────────────────────────────────────────
    async def test_connection(self) -> IntegrationTestResult:
        try:
            usage = await self.get_seat_usage()
        except Exception as e:  # noqa: BLE001
            return IntegrationTestResult(
                ok=False, message=f"{type(e).__name__}: {e}",
                detail={"type": type(e).__name__},
            )
        return IntegrationTestResult(
            ok=True,
            message=(
                f"OK · {usage.total_devices} managed device(s) "
                f"({usage.computers} Mac / {usage.mobile_devices} mobile)"
            ),
            detail={
                "computers": usage.computers,
                "mobile_devices": usage.mobile_devices,
                "total_devices": usage.total_devices,
            },
        )

    # ── seat usage (cheap totalCount probe) ──────────────────────
    async def _total_count(self, path: str) -> int:
        resp = await self._request(
            "GET", path, params=[("page", 0), ("page-size", 1)],
        )
        if resp.status_code != 200:
            raise JamfError(f"GET {path}", resp.status_code, resp.text[:200])
        return int(resp.json().get("totalCount", 0))

    async def get_seat_usage(self) -> MdmSeatUsage:
        computers = await self._total_count("/api/v1/computers-inventory")
        mobile = await self._total_count("/api/v2/mobile-devices")
        return MdmSeatUsage(
            total_devices=computers + mobile,
            computers=computers,
            mobile_devices=mobile,
        )

    # ── inventory ────────────────────────────────────────────────
    async def list_devices(self) -> list[MdmDevice]:
        devices = await self._list_computers()
        devices.extend(await self._list_mobile_devices())
        return devices

    async def get_device(
        self, *, device_type: str, external_id: str,
    ) -> MdmDevice | None:
        """Fetch a single device by inventory id (404 → None). Used by the
        offboarding action guard to resolve the device's serial before a
        destructive command."""
        if device_type == "computer":
            params: list[tuple[str, Any]] = [
                ("section", s) for s in _COMPUTER_SECTIONS
            ]
            resp = await self._request(
                "GET", f"/api/v1/computers-inventory-detail/{external_id}",
                params=params,
            )
            if resp.status_code == 404:
                return None
            if resp.status_code != 200:
                raise JamfError("get computer", resp.status_code, resp.text[:200])
            return self._computer_to_device(resp.json())
        resp = await self._request(
            "GET", f"/api/v2/mobile-devices/{external_id}/detail",
        )
        if resp.status_code == 404:
            return None
        if resp.status_code != 200:
            raise JamfError("get mobile device", resp.status_code, resp.text[:200])
        r = resp.json()
        loc = r.get("location") or {}
        return MdmDevice(
            external_id=str(r.get("mobileDeviceId") or external_id),
            device_type="mobile",
            management_id=r.get("managementId"),
            name=r.get("name"),
            serial_number=r.get("serialNumber"),
            model=r.get("model") or r.get("modelIdentifier"),
            os_version=r.get("osVersion"),
            managed=bool(r.get("managed", False)),
            assigned_user_email=(loc.get("emailAddress") or None),
            assigned_username=(loc.get("username") or None),
            raw=r,
        )

    async def _list_computers(self) -> list[MdmDevice]:
        out: list[MdmDevice] = []
        for page in range(_MAX_PAGES):
            params: list[tuple[str, Any]] = [
                ("page", page), ("page-size", _PAGE_SIZE),
                ("sort", "id:asc"),
            ]
            params.extend(("section", s) for s in _COMPUTER_SECTIONS)
            resp = await self._request(
                "GET", "/api/v1/computers-inventory", params=params,
            )
            if resp.status_code != 200:
                raise JamfError(
                    "list computers", resp.status_code, resp.text[:200],
                )
            results = resp.json().get("results", [])
            for r in results:
                out.append(self._computer_to_device(r))
            if len(results) < _PAGE_SIZE:
                break
        return out

    @staticmethod
    def _computer_to_device(r: dict[str, Any]) -> MdmDevice:
        general = r.get("general") or {}
        hardware = r.get("hardware") or {}
        os_info = r.get("operatingSystem") or {}
        ual = r.get("userAndLocation") or {}
        remote_mgmt = general.get("remoteManagement") or {}
        return MdmDevice(
            external_id=str(r.get("id", "")),
            device_type="computer",
            management_id=general.get("managementId"),
            name=general.get("name"),
            serial_number=hardware.get("serialNumber"),
            model=hardware.get("model") or hardware.get("modelIdentifier"),
            os_version=os_info.get("version"),
            managed=bool(remote_mgmt.get("managed", False)),
            assigned_user_email=(ual.get("email") or None),
            assigned_user_name=(ual.get("realname") or None),
            assigned_username=(ual.get("username") or None),
            last_contact_at=_parse_dt(general.get("lastContactTime")),
            raw=r,
        )

    async def _list_mobile_devices(self) -> list[MdmDevice]:
        out: list[MdmDevice] = []
        for page in range(_MAX_PAGES):
            resp = await self._request(
                "GET", "/api/v2/mobile-devices",
                params=[("page", page), ("page-size", _PAGE_SIZE)],
            )
            if resp.status_code != 200:
                raise JamfError(
                    "list mobile devices", resp.status_code, resp.text[:200],
                )
            results = resp.json().get("results", [])
            for r in results:
                out.append(
                    MdmDevice(
                        external_id=str(r.get("id", "")),
                        device_type="mobile",
                        management_id=r.get("managementId"),
                        name=r.get("name"),
                        serial_number=r.get("serialNumber"),
                        model=r.get("model") or r.get("modelIdentifier"),
                        os_version=r.get("osVersion"),
                        managed=bool(r.get("managed", False)),
                        assigned_username=(r.get("username") or None),
                        last_contact_at=_parse_dt(
                            r.get("lastInventoryUpdateTimestamp"),
                        ),
                        raw=r,
                    ),
                )
            if len(results) < _PAGE_SIZE:
                break
        return out

    # ── management-id resolution (commands key off this, not inv id) ─
    async def _management_id(
        self, device_type: str, external_id: str,
    ) -> str | None:
        """Resolve the device's ``managementId`` (commands key off this).

        Returns ``None`` when the device is GONE from Jamf inventory (404) —
        e.g. already erased + unenrolled — so a destructive-command caller
        can treat that as an idempotent no-op instead of hard-failing on a
        retry. Any OTHER non-200 raises.
        """
        if device_type == "computer":
            resp = await self._request(
                "GET", f"/api/v1/computers-inventory-detail/{external_id}",
                params=[("section", "GENERAL")],
            )
            if resp.status_code == 404:
                return None
            if resp.status_code != 200:
                raise JamfError(
                    "resolve managementId", resp.status_code, resp.text[:200],
                )
            mid = (resp.json().get("general") or {}).get("managementId")
        else:
            resp = await self._request(
                "GET", f"/api/v2/mobile-devices/{external_id}/detail",
            )
            if resp.status_code == 404:
                return None
            if resp.status_code != 200:
                raise JamfError(
                    "resolve managementId", resp.status_code, resp.text[:200],
                )
            mid = resp.json().get("managementId")
        if not mid:
            raise JamfError(
                "resolve managementId", None,
                f"no managementId for {device_type} {external_id}",
            )
        return str(mid)

    async def _send_command(
        self, device_type: str, external_id: str,
        command_data: dict[str, Any],
    ) -> MdmActionResult:
        management_id = await self._management_id(device_type, external_id)
        if management_id is None:
            # Device is gone from Jamf inventory (already erased/unenrolled).
            # The destructive command's goal is already met — idempotent
            # success (mirrors clear_device_assignment's 404 handling), so
            # an offboarding retry after a successful wipe doesn't hard-fail
            # and pin the leaver's blocker.
            return MdmActionResult(ok=True, already_done=True)
        body = {
            "clientData": [{"managementId": management_id}],
            "commandData": command_data,
        }
        # Destructive MDM commands have NO idempotency key — never blind-retry
        # a transient 5xx/429 (Jamf may have already queued the command), or
        # the device gets a double ERASE/LOCK.
        resp = await self._request(
            "POST", "/api/v2/mdm/commands", json_body=body,
            retry_on_transient=False,
        )
        if resp.status_code in (200, 201):
            data = resp.json()
            ref = None
            if isinstance(data, list) and data:
                ref = (data[0] or {}).get("id")
            return MdmActionResult(
                ok=True, external_ref=ref,
                raw_response={"results": data},
            )
        return MdmActionResult(
            ok=False,
            error_message=f"HTTP {resp.status_code}: {resp.text[:200]}",
            raw_response={"status": resp.status_code},
        )

    async def lock_device(
        self, *, device_type: str, external_id: str,
        message: str | None = None,
    ) -> MdmActionResult:
        command: dict[str, Any] = {"commandType": "DEVICE_LOCK"}
        if message:
            command["message"] = message
        return await self._send_command(device_type, external_id, command)

    async def wipe_device(
        self, *, device_type: str, external_id: str,
    ) -> MdmActionResult:
        return await self._send_command(
            device_type, external_id, {"commandType": "ERASE_DEVICE"},
        )

    async def clear_device_assignment(
        self, *, device_type: str, external_id: str,
    ) -> MdmActionResult:
        if device_type == "computer":
            path = f"/api/v1/computers-inventory-detail/{external_id}"
            body = {
                "userAndLocation": {
                    "username": "", "realname": "", "email": "",
                    "position": "", "phone": "",
                },
            }
        else:
            path = f"/api/v2/mobile-devices/{external_id}"
            body = {
                "location": {
                    "username": "", "realName": "", "emailAddress": "",
                    "position": "", "phoneNumber": "",
                },
            }
        resp = await self._request("PATCH", path, json_body=body)
        if resp.status_code in (200, 201):
            return MdmActionResult(ok=True, raw_response={"status": resp.status_code})
        if resp.status_code == 404:
            # Device gone == nothing to unbind. Idempotent success.
            return MdmActionResult(ok=True, already_done=True)
        return MdmActionResult(
            ok=False,
            error_message=f"HTTP {resp.status_code}: {resp.text[:200]}",
            raw_response={"status": resp.status_code},
        )

    async def set_device_assignment(
        self, *, device_type: str, external_id: str,
        email: str, realname: str | None = None, username: str | None = None,
    ) -> MdmActionResult:
        """Write the assigned user onto a device's User-and-Location record
        (the inverse of :meth:`clear_device_assignment`). Reversible metadata
        edit — does NOT touch the physical device. Used to backfill Jamf
        ownership from Helix's asset system when Jamf has no assigned user."""
        if device_type == "computer":
            path = f"/api/v1/computers-inventory-detail/{external_id}"
            body = {
                "userAndLocation": {
                    "username": username or email,
                    "realname": realname or "",
                    "email": email,
                },
            }
        else:
            path = f"/api/v2/mobile-devices/{external_id}"
            body = {
                "location": {
                    "username": username or email,
                    "realName": realname or "",
                    "emailAddress": email,
                },
            }
        resp = await self._request("PATCH", path, json_body=body)
        if resp.status_code in (200, 201):
            return MdmActionResult(ok=True, raw_response={"status": resp.status_code})
        if resp.status_code == 404:
            return MdmActionResult(
                ok=False, error_message="device not found",
                raw_response={"status": 404},
            )
        return MdmActionResult(
            ok=False,
            error_message=f"HTTP {resp.status_code}: {resp.text[:200]}",
            raw_response={"status": resp.status_code},
        )