"""NexusCRM software-provisioning driver.

NexusCRM (``nexuscrm-new-prod.pingcap.com``) exposes a
Salesforce-compatible REST API: ``/services/data/vNN.0`` with SOQL
``/query``, ``sobjects/User`` records, and the OAuth 2.0
``client_credentials`` flow via an "IT Connected App" (Consumer Key /
Consumer Secret). A "seat" is an **active** ``User`` record
(``IsActive = true``); the platform can't hard-delete a CRM user, so
"clearing a seat" means flipping ``IsActive`` to ``false`` — which
frees the license and is reversible by re-activating.

Auth:
  POST {instance}/services/oauth2/token
       grant_type=client_credentials
       client_id=<consumer_key> client_secret=<consumer_secret>
  → {"access_token": "<bearer>", "instance_url": ...}

The Connected App's run-as user determines what the token can do. A
read-only run-as user (scope ``api.read``) lists seats fine but every
write returns ``INSUFFICIENT_SCOPE`` 403 — the driver surfaces that as
a clear, non-transient ``ProvisioningResult.ok=False`` so an admin
knows the Connected App needs ``api`` (write) scope, rather than
silently confirming a revoke that never happened.

Offboarding for NexusCRM runs ``manual_notify`` (see the seeded
connector binding) — the leaver's seat-holder owner is DM'd; the
driver's ``revoke_seat`` is NOT called on the offboarding path. It IS
called by the one-time seat-cleanup script and by any future
``auto_api`` binding.
"""
from __future__ import annotations

import asyncio
import logging
import re
import time
from typing import Any

import httpx

from app.core.pagination import PaginationGuard
from app.integrations.driver_base import (
    ProvisioningContext,
    ProvisioningResult,
    SoftwareProvisioningDriver,
)
from app.integrations.registry import register_driver
from app.integrations.schemas import (
    IntegrationFieldInfo,
    IntegrationTestResult,
)

logger = logging.getLogger(__name__)

_DEFAULT_API_VERSION = "v63.0"
_HTTP_TIMEOUT = httpx.Timeout(30.0, connect=10.0)
# Refresh the cached token this many seconds before its stated expiry so
# a long-running sync doesn't fire a request with a token that expires
# mid-flight.
_TOKEN_EXPIRY_SKEW = 60
_MAX_PAGINATION_PAGES = 1000
_MAX_PAGINATION_ITEMS = 2_000_000


def _soql_str(value: str) -> str:
    """Escape a string literal for inclusion in a SOQL ``WHERE`` clause.

    SOQL (like SQL) single-quotes string literals and backslash-escapes
    embedded quotes/backslashes. The ``work_email`` we filter on is
    HRIS-controlled in normal operation, but multi-vendor sync paths can
    feed it externally — defence-in-depth so a poisoned value like
    ``a' OR Name != '`` can't break out of the literal. Mirrors the
    SCIM-filter escaping in github_software.py.
    """
    return value.replace("\\", "\\\\").replace("'", "\\'")


# Salesforce platform / integration / synthetic profile-name fragments.
# These accounts hold special license types (not paid seats), several
# cannot be deactivated at all, and the Connected App's own run-as user
# lives here — the seat-cleanup script MUST skip every one of them.
_SYSTEM_PROFILE_FRAGMENTS = (
    "Integration User",        # Analytics Cloud / Platform Integration User
    "Automated Process",
    "Chatter",                 # Chatter Expert / Chatter Free
    "Identity User",
)


def _is_system_account(email: str, profile: str | None) -> bool:
    """True for Salesforce platform / integration / synthetic accounts.

    Detection keys off two signals (this instance's ``User`` has no
    ``UserType`` column):

      * **email domain** — real employees are ``@<corp domain>``; platform
        accounts use the 15/18-char Salesforce org id as the domain
        (``cloud@00d5g000002g54ueac``), ``@salesforce.com``, or a synthetic
        ``.invalid`` placeholder from a data migration.
      * **profile name** — ``*Integration User``, ``Automated Process``,
        ``Chatter*`` etc.
    """
    e = (email or "").strip().lower()
    domain = e.rsplit("@", 1)[-1] if "@" in e else ""
    if (
        domain.startswith("00d")          # Salesforce org-id domain
        or domain == "salesforce.com"
        or domain.endswith(".invalid")    # migration placeholder
    ):
        return True
    p = profile or ""
    return any(frag in p for frag in _SYSTEM_PROFILE_FRAGMENTS)


def _redact_secrets(blob: str, *known_secrets: str) -> str:
    """Strip bearer tokens / consumer secrets from error-snippet blobs
    before they're persisted into ``provisioning_tasks.error_message``
    or ``software_connectors.last_error_message`` (both admin-visible)."""
    if not blob:
        return blob
    for secret in known_secrets:
        if secret:
            blob = blob.replace(secret, "***")
    blob = re.sub(
        r"(?i)(authorization\s*:?\s*bearer\s+)[A-Za-z0-9._\-+/=]+",
        r"\1***",
        blob,
    )
    # Salesforce session/JWT access tokens are long dot-or-bang-delimited
    # opaque strings; redact anything that looks like one in a snippet.
    blob = re.sub(r"\beyJ[A-Za-z0-9._\-]{20,}", "eyJ***", blob)
    return blob


@register_driver
class NexusCrmSoftwareDriver(SoftwareProvisioningDriver):
    provider = "nexuscrm"
    display_name = "NexusCRM"
    description = (
        "NexusCRM(Salesforce 兼容 REST API)。通过 IT Connected App 的 "
        "OAuth client_credentials(Consumer Key / Consumer Secret)鉴权。"
        "坐席 = IsActive=true 的 User 记录;revoke = 把 User 置为 "
        "IsActive=false(释放许可,可逆)。需要 Connected App 的 run-as "
        "用户具备 api(写)scope 才能 revoke;只有 api.read 时列表正常、"
        "写操作返回 INSUFFICIENT_SCOPE。"
    )
    supports_seat = True
    supports_permission = False
    supports_multi_instance = True
    # Deactivating the User strips the user's entire CRM footprint, so the
    # single-seat revoke IS the full-access revoke (no group model on top).
    supports_full_access_revoke = True

    config_schema = [
        IntegrationFieldInfo(
            key="instance_url", label="Instance URL",
            is_secret=False, required=True,
            placeholder="https://nexuscrm-new-prod.pingcap.com",
            hint="NexusCRM 实例地址(REST API 根 /services/data 所在的域)。",
        ),
        IntegrationFieldInfo(
            key="consumer_key", label="Consumer Key",
            is_secret=False, required=True,
            placeholder="Connected App 的 Consumer Key (client_id)",
            hint="IT Connected App 的 Consumer Key。",
        ),
        IntegrationFieldInfo(
            key="consumer_secret", label="Consumer Secret",
            is_secret=True, required=True,
            placeholder="Connected App 的 Consumer Secret (client_secret)",
            hint="IT Connected App 的 Consumer Secret。写操作需要 run-as "
                 "用户带 api scope。",
        ),
        IntegrationFieldInfo(
            key="api_version", label="API Version",
            is_secret=False, required=False,
            default=_DEFAULT_API_VERSION,
            placeholder=_DEFAULT_API_VERSION,
            hint="REST API 版本,通常留空用默认(v63.0)。",
        ),
    ]
    binding_schema: list[IntegrationFieldInfo] = []  # org-level; no per-software params

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        # Per-instance token cache. Driver lifetime is per-request, but a
        # single seat-sync makes many query calls — fetch the token once.
        self._token: str | None = None
        self._token_expires_at: float = 0.0
        self._safe_target: Any | None = None
        self._safe_target_url: str | None = None
        self._safe_target_lock = asyncio.Lock()

    # ── Config accessors ─────────────────────────────────────────

    def _instance_url(self) -> str:
        url = str(self.config.get("instance_url") or "").rstrip("/")
        if not url:
            raise ValueError("NexusCRM integration missing instance_url")
        return url

    async def _transport(self):
        from app.integrations._url_safety import (
            PinnedAsyncHTTPTransport,
            resolve_safe_external_url_async,
        )
        url = self._instance_url()
        if self._safe_target_url != url:
            async with self._safe_target_lock:
                if self._safe_target_url != url:
                    target = await resolve_safe_external_url_async(url)
                    self._safe_target = target
                    self._safe_target_url = url
        if self._safe_target is None:
            raise RuntimeError("NexusCRM safe URL target was not initialized")
        return PinnedAsyncHTTPTransport(self._safe_target)

    def _api_version(self) -> str:
        return str(self.config.get("api_version") or _DEFAULT_API_VERSION).strip()

    def _consumer_key(self) -> str:
        v = self.config.get("consumer_key")
        if not v:
            raise ValueError("NexusCRM integration missing consumer_key")
        return str(v)

    def _consumer_secret(self) -> str:
        v = self.config.get("consumer_secret")
        if not v:
            raise ValueError("NexusCRM integration missing consumer_secret")
        return str(v)

    def _data_base(self) -> str:
        return f"{self._instance_url()}/services/data/{self._api_version()}"

    # ── OAuth ────────────────────────────────────────────────────

    async def _fetch_token(self, client: httpx.AsyncClient) -> str:
        """Run the client_credentials flow and cache the bearer token."""
        now = time.monotonic()
        if self._token is not None and now < self._token_expires_at:
            return self._token
        resp = await client.post(
            f"{self._instance_url()}/services/oauth2/token",
            data={
                "grant_type": "client_credentials",
                "client_id": self._consumer_key(),
                "client_secret": self._consumer_secret(),
            },
        )
        if resp.status_code != 200:
            raise RuntimeError(
                f"NexusCRM OAuth token request failed {resp.status_code}: "
                f"{_redact_secrets(resp.text[:200], self._consumer_secret())}"
            )
        data = resp.json() or {}
        token = data.get("access_token")
        if not token:
            raise RuntimeError(
                "NexusCRM OAuth response had no access_token: "
                f"{_redact_secrets(str(data)[:200], self._consumer_secret())}"
            )
        # The token JWT carries ``exp`` but we don't decode it here; tokens
        # are short-lived (≈24h observed) and a single sync finishes in
        # seconds. Cache conservatively for 15 minutes within this driver
        # instance's lifetime.
        self._token = str(token)
        self._token_expires_at = now + 900 - _TOKEN_EXPIRY_SKEW
        return self._token

    async def _bearer_headers(self, client: httpx.AsyncClient) -> dict[str, str]:
        return {"Authorization": f"Bearer {await self._fetch_token(client)}"}

    # ── SOQL query (paginated) ───────────────────────────────────

    async def _query_all(
        self, client: httpx.AsyncClient, soql: str,
    ) -> list[dict[str, Any]]:
        """Run a SOQL query and follow ``nextRecordsUrl`` pagination."""
        headers = await self._bearer_headers(client)
        records: list[dict[str, Any]] = []
        guard = PaginationGuard(
            integration="nexuscrm",
            action="query_all",
            max_pages=_MAX_PAGINATION_PAGES,
            max_items=_MAX_PAGINATION_ITEMS,
        )
        resp = await client.get(
            f"{self._data_base()}/query",
            headers=headers,
            params={"q": soql},
        )
        while True:
            if resp.status_code != 200:
                body_preview = _redact_secrets(
                    resp.text[:200],
                    self._consumer_secret(),
                    self._token or "",
                )
                raise RuntimeError(
                    f"NexusCRM query failed {resp.status_code}: {body_preview}"
                )
            data = resp.json()
            if not isinstance(data, dict):
                guard.reject_malformed_response()
            page_records = data.get("records")
            done = data.get("done")
            if (
                not isinstance(page_records, list)
                or any(not isinstance(item, dict) for item in page_records)
                or not isinstance(done, bool)
            ):
                guard.reject_malformed_response()
            guard.record_page(len(page_records))
            records.extend(page_records)
            next_url = data.get("nextRecordsUrl")
            if next_url is not None and not isinstance(next_url, str):
                guard.reject_malformed_response()
            if done:
                break
            guard.require_safe_next(next_url, required=True)
            resp = await client.get(
                f"{self._instance_url()}{next_url}", headers=headers,
            )
        return records

    # ── test_connection ──────────────────────────────────────────

    async def test_connection(self) -> IntegrationTestResult:
        """Probe: token exchange + count active Users."""
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                rows = await self._query_all(
                    client,
                    "SELECT COUNT(Id) total FROM User WHERE IsActive = true",
                )
        except httpx.HTTPError as e:
            return IntegrationTestResult(
                ok=False,
                message=_redact_secrets(
                    f"{type(e).__name__}: {e}",
                    self._consumer_secret(),
                    self._token or "",
                ),
            )
        except Exception as e:  # noqa: BLE001
            return IntegrationTestResult(
                ok=False,
                message=_redact_secrets(
                    f"{type(e).__name__}: {e}",
                    self._consumer_secret(),
                    self._token or "",
                ),
            )
        active = 0
        if rows:
            # Aggregate query returns one row with the alias.
            active = int(rows[0].get("total") or rows[0].get("expr0") or 0)
        return IntegrationTestResult(
            ok=True,
            message=f"OK · {active} active seat(s)",
            detail={"active_seats": active, "instance": self._instance_url()},
        )

    # ── list seats (reconciliation source) ───────────────────────

    async def list_full_members(self) -> list[dict[str, Any]]:
        """Every active CRM User (a "seat"), for ``nexuscrm.seat_sync``.

        Returns one dict per active user with the fields the reconciliation
        handler needs to match against ``identity_persons`` and the
        seat-cleanup script needs to exclude platform accounts:

          ``id`` · ``email`` · ``display_name`` · ``profile`` ·
          ``is_system``

        ``is_system`` flags Salesforce platform / integration / synthetic
        accounts (the Connected App's own run-as user, Automated Process,
        Chatter Expert, Data.com Clean, Insights Integration, migration
        placeholders) — wiping any of those would break the connector or
        the platform, so the cleanup script must skip them. Note: this
        instance's ``User`` object exposes no ``UserType`` column, so
        detection keys off ``Profile.Name`` + the account's email domain.
        """
        async with httpx.AsyncClient(
            timeout=_HTTP_TIMEOUT,
            trust_env=False,
            transport=await self._transport(),
        ) as client:
            rows = await self._query_all(
                client,
                # ``ProfileId`` must be selected for this instance to
                # populate the nested ``Profile`` relationship object.
                "SELECT Id, Name, Email, IsActive, ProfileId, Profile.Name "
                "FROM User WHERE IsActive = true",
            )
        members: list[dict[str, Any]] = []
        for r in rows:
            profile = (
                (r.get("Profile") or {}).get("Name")
                if isinstance(r.get("Profile"), dict)
                else None
            ) or r.get("ProfileId_Name")
            email = (r.get("Email") or "").strip()
            members.append({
                "id": r.get("Id"),
                "email": email,
                "display_name": r.get("Name"),
                "profile": profile,
                "is_system": _is_system_account(email, profile),
            })
        return members

    # ── user lookup ──────────────────────────────────────────────

    async def _find_active_user(
        self, client: httpx.AsyncClient, email: str,
    ) -> tuple[str | None, bool, str | None]:
        """Look up a User id by email.

        Returns ``(user_id, is_active, network_error)``:
          * ``user_id`` is ``None`` when no User matches the email.
          * ``network_error`` is a short label when the lookup failed at
            the transport / API layer; callers MUST NOT interpret this as
            "user already absent" (that would silently confirm a revoke
            against a still-active seat — the github_software.py lesson).
        """
        try:
            rows = await self._query_all(
                client,
                f"SELECT Id, IsActive FROM User WHERE Email = '{_soql_str(email)}'",
            )
        except Exception as e:  # noqa: BLE001
            return None, False, _redact_secrets(
                f"{type(e).__name__}: {e}",
                self._consumer_secret(),
                self._token or "",
            )
        if not rows:
            return None, False, None
        # Salesforce ``User.Email`` is NOT unique — a person can have a stale
        # inactive duplicate alongside a live account. Prefer the ACTIVE row;
        # returning an inactive dupe (which may sort first, no ORDER BY) would
        # let ``_deactivate`` report ``already_inactive`` success and close the
        # offboarding blocker while the live billable seat stays active (the
        # github_software.py silent-confirm-against-active-seat lesson).
        chosen = next((r for r in rows if bool(r.get("IsActive"))), rows[0])
        return str(chosen.get("Id")), bool(chosen.get("IsActive")), None

    # ── revoke ───────────────────────────────────────────────────

    async def revoke_seat(self, ctx: ProvisioningContext) -> ProvisioningResult:
        """Deactivate the CRM User (frees the seat). Idempotent.

        A missing user, or one already ``IsActive=false``, is treated as
        success so an offboarding/cleanup re-run closes cleanly.
        """
        return await self._deactivate(ctx)

    async def revoke_all_access(self, ctx: ProvisioningContext) -> ProvisioningResult:
        """No group model on top of the User — deactivating IS the full
        revoke."""
        return await self._deactivate(ctx)

    async def _deactivate(self, ctx: ProvisioningContext) -> ProvisioningResult:
        try:
            async with httpx.AsyncClient(
                timeout=_HTTP_TIMEOUT,
                trust_env=False,
                transport=await self._transport(),
            ) as client:
                user_id, is_active, net_err = await self._find_active_user(
                    client, ctx.user_email,
                )
                if net_err is not None:
                    return ProvisioningResult(
                        ok=False,
                        error_message=(
                            f"NexusCRM user lookup transport/API error for "
                            f"{ctx.user_email!r}: {net_err}"
                        ),
                    )
                if user_id is None:
                    return ProvisioningResult(
                        ok=True,
                        external_ref=f"nexuscrm:{ctx.user_email}:already_absent",
                        raw_response={"already_absent": True},
                    )
                if not is_active:
                    return ProvisioningResult(
                        ok=True,
                        external_ref=user_id,
                        raw_response={"already_inactive": True},
                    )
                headers = await self._bearer_headers(client)
                headers["Content-Type"] = "application/json"
                resp = await client.patch(
                    f"{self._data_base()}/sobjects/User/{user_id}",
                    headers=headers,
                    json={"IsActive": False},
                )
        except httpx.HTTPError as e:
            return ProvisioningResult(
                ok=False,
                error_message=_redact_secrets(
                    f"{type(e).__name__}: {e}",
                    self._consumer_secret(),
                    self._token or "",
                ),
            )

        if resp.status_code in (200, 204):
            return ProvisioningResult(
                ok=True, external_ref=user_id, raw_response={"deactivated": True},
            )
        body_preview = _redact_secrets(
            resp.text[:300], self._consumer_secret(), self._token or "",
        )
        # INSUFFICIENT_SCOPE = the Connected App's run-as user is read-only
        # (api.read). This is a real configuration gap, not a transient
        # failure — make the message actionable instead of silently
        # retrying forever.
        if resp.status_code == 403 and "INSUFFICIENT_SCOPE" in resp.text:
            return ProvisioningResult(
                ok=False,
                error_message=(
                    f"NexusCRM revoke of {ctx.user_email!r} blocked: the IT "
                    "Connected App token is read-only (api.read). Grant the "
                    "Connected App 'api' (write) scope and give its "
                    "client-credentials run-as user permission to edit Users, "
                    "then retry."
                ),
                raw_response={"status": 403, "body": body_preview},
            )
        return ProvisioningResult(
            ok=False,
            error_message=f"NexusCRM deactivate {resp.status_code}: {body_preview}",
            raw_response={"status": resp.status_code, "body": body_preview},
        )