#!/usr/bin/env python3
"""Apply the Catalog-owned PingCAP Connected App Seed module.

Seeds the reviewed ``_System_ConnectedApp`` clients plus the ``User`` each runs
as, so both connected apps are preserved across
new-staging / EXP deployment resets. The run-as User is get-or-create matched
on ``Username``: created when absent (fresh EXP), reused without any PATCH when
already present (new-staging restored from the prod branch baseline), after its
seed-declared binding fields are verified so authorization drift is never
silently accepted and the real user's contact fields are never clobbered.
"""

from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import sys
import time
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import (
    HTTPRedirectHandler,
    ProxyHandler,
    Request,
    build_opener,
    urlopen,
)


SCRIPT_DIR = Path(__file__).resolve().parent
REPO_ROOT = SCRIPT_DIR.parent
RUNTIME_DIR = REPO_ROOT / "deployments" / "offline" / "runtime"
if str(RUNTIME_DIR) not in sys.path:
    sys.path.insert(0, str(RUNTIME_DIR))

import deploy_offline_env  # noqa: E402


MODULE_ID = "connected_apps"
SUMMARY_FILENAME = "summary.json"
SUMMARY_FORMAT = "deploy-seed-script-v1"
PLANNED_COUNTS = {"connected_apps": 2, "seed_users": 2}
API_VERSION = "v63.0"
USER_CREATE_FIELDS = (
    "Username",
    "Email",
    "FirstName",
    "LastName",
    "Alias",
    "ProfileId",
    "IsActive",
)
USER_BINDING_FIELDS = (
    "Username",
    "ProfileId",
    "IsActive",
)
CONNECTED_APP_FIELDS = (
    "Name",
    "ConsumerKey",
    "ConsumerSecret",
    "IsActive",
    "Scopes",
    "PermittedUsers",
    "RunAsUserId",
)
CONNECTED_APP_CREATE_ONLY_FIELDS = frozenset({"ConsumerKey", "ConsumerSecret"})
PROFILE_DISPLAY_NAMES = {
    "system_admin": "System Administrator",
}


class ConnectedAppsSeedAdapterError(RuntimeError):
    """Raised when the scripted Seed adapter contract is violated."""


class NoRedirectHandler(HTTPRedirectHandler):
    """Keep verification bearer tokens on the exact local D3 origin."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: ANN001
        return None


def open_verification_request(request: Request, *, timeout: int):
    """Open a local verification request without proxies or redirects."""
    return build_opener(ProxyHandler({}), NoRedirectHandler()).open(
        request,
        timeout=timeout,
    )


def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Apply PingCAP Connected App Seed data.")
    parser.add_argument("--source-root", required=True)
    parser.add_argument("--module-id", required=True)
    parser.add_argument("--invocation-id", required=True)
    parser.add_argument("--data-source", action="append", required=True)
    parser.add_argument("--selection-json", required=True)
    parser.add_argument("--output-dir", required=True)
    parser.add_argument("--dry-run", action="store_true")
    return parser.parse_args(argv)


def _selection(raw: str) -> dict[str, Any]:
    try:
        selection = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ConnectedAppsSeedAdapterError("selection must be valid JSON") from exc
    if selection != {}:
        raise ConnectedAppsSeedAdapterError(
            "PingCAP Connected App Seed does not accept a selection"
        )
    return selection


def _seed_file(source_root: Path, data_sources: list[str]) -> Path:
    if len(data_sources) != 1:
        raise ConnectedAppsSeedAdapterError(
            "PingCAP Connected App Seed requires one data source"
        )
    expected = (
        source_root
        / "deployments"
        / "offline"
        / "runtime"
        / "seeds"
        / "connected-apps.seed.yaml"
    ).resolve(strict=True)
    actual = Path(data_sources[0]).expanduser().resolve(strict=True)
    if actual != expected:
        raise ConnectedAppsSeedAdapterError(
            "data source is not the reviewed Connected App Seed"
        )
    return actual


def _record_ref_key(reference: Any, *, field: str) -> str:
    expression = str(reference).strip()
    if not expression.startswith("{{") or not expression.endswith("}}"):
        raise ConnectedAppsSeedAdapterError(
            f"Connected App Seed {field} must be a record ref, not {reference!r}"
        )
    expression = expression.removeprefix("{{").removesuffix("}}").strip()
    if not expression.startswith("record.") or not expression.endswith(".id"):
        raise ConnectedAppsSeedAdapterError(
            f"Connected App Seed {field} must be a record ref, not {reference!r}"
        )
    key = expression.removeprefix("record.").removesuffix(".id").strip()
    if not key:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App Seed {field} has an empty record ref"
        )
    return key


def _resolve_env_ref(value: Any, *, field: str, environment: dict[str, str]) -> Any:
    """Resolve a ``{{ env.KEY }}`` field value from the deploy environment.

    Secrets (e.g. ConsumerSecret) are never tracked literals; they arrive at
    apply time via the deploy environment. Non-env templates (e.g. a record
    ref) are returned unchanged for resolution elsewhere.
    """
    text = str(value).strip()
    if not (text.startswith("{{") and text.endswith("}}")):
        return value
    inner = text.removeprefix("{{").removesuffix("}}").strip()
    if not inner.startswith("env."):
        return value
    key = inner.removeprefix("env.").strip()
    if not key:
        raise ConnectedAppsSeedAdapterError(f"Connected App {field} has an empty env ref")
    resolved = str(environment.get(key, "") or "").strip()
    if not resolved:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App {field} requires environment variable {key!r} to be set"
        )
    return resolved


def load_connected_apps(seed_file: Path) -> dict[str, Any]:
    sections = deploy_offline_env.load_simple_seed_sections(
        seed_file, {"lookup_records", "baseline_records"}
    )
    records = sections["baseline_records"]
    if not records:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App Seed has no baseline_records: {seed_file}"
        )

    users: dict[str, dict[str, Any]] = {}
    connected_apps: list[dict[str, Any]] = []
    for index, record in enumerate(records, start=1):
        if not isinstance(record, dict):
            raise ConnectedAppsSeedAdapterError(
                f"Connected App Seed row {index} must be an object"
            )
        obj = record.get("object")
        match = record.get("match")
        fields = record.get("fields")
        key = str(record.get("key") or "").strip()
        if not key or not isinstance(match, dict) or not isinstance(fields, dict):
            raise ConnectedAppsSeedAdapterError(
                f"Connected App Seed row {index} missing key/match/fields"
            )
        if obj == "User":
            username = str(match.get("Username") or "").strip()
            if not username:
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App Seed User row {index} must match on Username"
                )
            create_fields = {
                name: fields[name] for name in USER_CREATE_FIELDS if name in fields
            }
            missing_user_fields = [
                name for name in USER_CREATE_FIELDS if name not in create_fields
            ]
            if missing_user_fields:
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App Seed User row {index} missing fields: "
                    f"{missing_user_fields}"
                )
            if str(create_fields.get("Username") or "").strip() != username:
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App Seed User row {index} Username field must equal its match"
                )
            existing_profile_name = str(
                match.get("ExistingProfileName") or ""
            ).strip()
            if not existing_profile_name:
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App Seed User row {index} has empty "
                    "match.ExistingProfileName"
                )
            users[key] = {
                "username": username,
                "create_fields": create_fields,
                "existing_profile_name": existing_profile_name,
            }
        elif obj == "_System_ConnectedApp":
            consumer_key = str(match.get("ConsumerKey") or "").strip()
            if not consumer_key:
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App Seed row {index} must match on ConsumerKey"
                )
            missing = [name for name in CONNECTED_APP_FIELDS if name not in fields]
            if missing:
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App Seed row {index} missing fields: {missing}"
                )
            run_as_key = _record_ref_key(fields["RunAsUserId"], field="RunAsUserId")
            connected_apps.append(
                {
                    "consumer_key": consumer_key,
                    "run_as_user_key": run_as_key,
                    "fields": {name: fields[name] for name in CONNECTED_APP_FIELDS},
                }
            )
        else:
            raise ConnectedAppsSeedAdapterError(
                f"Connected App Seed row {index} has unsupported object {obj!r}"
            )

    if len(users) != PLANNED_COUNTS["seed_users"]:
        raise ConnectedAppsSeedAdapterError(
            "Connected App Seed must define exactly "
            f"{PLANNED_COUNTS['seed_users']} run-as User row(s); found {len(users)}"
        )
    if len(connected_apps) != PLANNED_COUNTS["connected_apps"]:
        raise ConnectedAppsSeedAdapterError(
            "Connected App Seed must define exactly "
            f"{PLANNED_COUNTS['connected_apps']} _System_ConnectedApp row(s); "
            f"found {len(connected_apps)}"
        )
    for app in connected_apps:
        if app["run_as_user_key"] not in users:
            raise ConnectedAppsSeedAdapterError(
                f"Connected App run-as ref {app['run_as_user_key']!r} has no seeded User row"
            )
    return {"users": users, "connected_apps": connected_apps}


def _normalized_boolean(value: Any, *, label: str) -> bool:
    if isinstance(value, bool):
        return value
    normalized = str(value).strip().lower()
    if normalized in {"1", "true", "yes", "on"}:
        return True
    if normalized in {"0", "false", "no", "off"}:
        return False
    raise ConnectedAppsSeedAdapterError(
        f"Connected App run-as User {label} is not boolean: {value!r}"
    )


def _resolve_runtime_profile_id(
    base_url: str,
    access_token: str,
    *,
    seed_profile_id: Any,
) -> str:
    profile_key = str(seed_profile_id or "").strip()
    profile_name = PROFILE_DISPLAY_NAMES.get(profile_key)
    if profile_name is None:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User has unsupported ProfileId key {profile_key!r}"
        )
    records = deploy_offline_env.exp_soql_records(
        base_url,
        access_token,
        "SELECT Id, Name FROM Profile WHERE Name = "
        f"'{deploy_offline_env.soql_string_literal(profile_name)}'",
    )
    if len(records) != 1:
        raise ConnectedAppsSeedAdapterError(
            "Connected App run-as User profile resolution requires exactly one "
            f"Profile named {profile_name!r}; found {len(records)}"
        )
    profile_id = str(records[0].get("Id") or "").strip()
    if not profile_id:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User Profile {profile_name!r} is missing Id"
        )
    return profile_id


def _resolve_runtime_user_fields(
    base_url: str,
    access_token: str,
    *,
    seed_fields: dict[str, Any],
) -> dict[str, Any]:
    resolved = dict(seed_fields)
    resolved["ProfileId"] = _resolve_runtime_profile_id(
        base_url,
        access_token,
        seed_profile_id=seed_fields.get("ProfileId"),
    )
    return resolved


def _profile_display_name(profile_key: Any) -> str:
    normalized = str(profile_key or "").strip()
    profile_name = PROFILE_DISPLAY_NAMES.get(normalized)
    if profile_name is None:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User has unsupported ProfileId key {normalized!r}"
        )
    return profile_name


def _require_user_matches_seed(
    record: dict[str, Any],
    *,
    username: str,
    expected_fields: dict[str, Any],
    expected_profile_name: str,
    existing_profile_name: str,
) -> str:
    user_id = str(record.get("Id") or "").strip()
    if not user_id:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User lookup missing Id for {username!r}"
        )
    for field in ("Username", "IsActive"):
        expected = expected_fields[field]
        actual = record.get(field)
        if field == "IsActive":
            matches = _normalized_boolean(
                actual,
                label=f"{field} readback for {username!r}",
            ) == _normalized_boolean(
                expected,
                label=f"expected {field} for {username!r}",
            )
        else:
            matches = str(actual or "").strip() == str(expected or "").strip()
        if not matches:
            raise ConnectedAppsSeedAdapterError(
                f"Connected App run-as User {field} mismatch for {username!r}: "
                f"expected {expected!r}, found {actual!r}"
            )

    expected_profile_id = str(expected_fields.get("ProfileId") or "").strip()
    actual_profile_id = str(record.get("ProfileId") or "").strip()
    profile = record.get("Profile")
    actual_profile_name = ""
    if isinstance(profile, dict):
        actual_profile_name = str(profile.get("Name") or "").strip()
    if not actual_profile_name:
        actual_profile_name = str(
            record.get("ProfileId_Name") or record.get("Profile.Name") or ""
        ).strip()

    matches_create_profile = actual_profile_id == expected_profile_id
    if matches_create_profile and actual_profile_name:
        matches_create_profile = actual_profile_name == expected_profile_name
    matches_reviewed_existing_profile = actual_profile_name == existing_profile_name
    if not (matches_create_profile or matches_reviewed_existing_profile):
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User Profile mismatch for {username!r}: "
            f"expected create ProfileId {expected_profile_id!r} "
            f"({expected_profile_name!r}) or existing Profile.Name "
            f"{existing_profile_name!r}, found ProfileId {actual_profile_id!r} "
            f"({actual_profile_name!r})"
        )
    return user_id


def _query_users_by_username(
    base_url: str,
    access_token: str,
    *,
    username: str,
) -> list[dict[str, Any]]:
    select_fields = ", ".join(("Id", *USER_BINDING_FIELDS, "Profile.Name"))
    return deploy_offline_env.exp_soql_records(
        base_url,
        access_token,
        f"SELECT {select_fields} FROM User WHERE Username = "
        f"'{deploy_offline_env.soql_string_literal(username)}'",
    )


def _get_or_create_user(
    base_url: str,
    access_token: str,
    *,
    username: str,
    create_fields: dict[str, Any],
    existing_profile_name: str,
) -> dict[str, str]:
    runtime_fields = _resolve_runtime_user_fields(
        base_url,
        access_token,
        seed_fields=create_fields,
    )
    expected_profile_name = _profile_display_name(create_fields.get("ProfileId"))
    records = _query_users_by_username(
        base_url,
        access_token,
        username=username,
    )
    if len(records) > 1:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User lookup found duplicate rows for {username!r}: "
            f"{len(records)}"
        )
    if records:
        # Existing user (e.g. new-staging via prod branch-reset): reuse the Id
        # only after proving its selector, profile, and active state. Never
        # PATCH the real user; authorization drift fails before the Connected
        # App is written.
        user_id = _require_user_matches_seed(
            records[0],
            username=username,
            expected_fields=runtime_fields,
            expected_profile_name=expected_profile_name,
            existing_profile_name=existing_profile_name,
        )
        return {"id": user_id, "action": "matched"}

    created = deploy_offline_env.api_json_request(
        base_url,
        f"/services/data/{API_VERSION}/sobjects/User",
        access_token=access_token,
        method="POST",
        payload=runtime_fields,
    )
    user_id = str(created.get("id") or created.get("Id") or "").strip()
    if not user_id:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App run-as User create missing id for {username!r}"
        )
    return {"id": user_id, "action": "created"}


def _lookup_connected_app(
    base_url: str,
    access_token: str,
    *,
    consumer_key: str,
) -> str | None:
    records = deploy_offline_env.exp_soql_records(
        base_url,
        access_token,
        "SELECT Id FROM _System_ConnectedApp WHERE ConsumerKey = "
        f"'{deploy_offline_env.soql_string_literal(consumer_key)}'",
    )
    if len(records) > 1:
        raise ConnectedAppsSeedAdapterError(
            "Connected App lookup found duplicate rows for ConsumerKey "
            f"{consumer_key!r}: {len(records)}"
        )
    if records:
        record_id = str(records[0].get("Id") or "").strip()
        if not record_id:
            raise ConnectedAppsSeedAdapterError(
                f"Connected App lookup missing Id for ConsumerKey {consumer_key!r}"
            )
        return record_id
    return None


def _upsert_connected_app(
    base_url: str,
    access_token: str,
    *,
    consumer_key: str,
    existing_record_id: str | None,
    fields: dict[str, Any],
) -> dict[str, str]:
    if existing_record_id is not None:
        # ConsumerKey identifies the existing app and is immutable after
        # creation. The supplied secret was already proved against this exact
        # app before any seed mutation; preserve its non-readable stored hash
        # and never turn an unverified runner value into a credential rotation.
        update_fields = {
            name: value
            for name, value in fields.items()
            if name not in CONNECTED_APP_CREATE_ONLY_FIELDS
        }
        deploy_offline_env.api_json_request(
            base_url,
            f"/services/data/{API_VERSION}/sobjects/_System_ConnectedApp/"
            f"{deploy_offline_env.quote(existing_record_id, safe='')}",
            access_token=access_token,
            method="PATCH",
            payload=update_fields,
        )
        return {"id": existing_record_id, "action": "updated"}

    created = deploy_offline_env.api_json_request(
        base_url,
        f"/services/data/{API_VERSION}/sobjects/_System_ConnectedApp",
        access_token=access_token,
        method="POST",
        payload=fields,
    )
    record_id = str(created.get("id") or created.get("Id") or "").strip()
    if not record_id:
        raise ConnectedAppsSeedAdapterError(
            f"Connected App create missing id for ConsumerKey {consumer_key!r}"
        )
    return {"id": record_id, "action": "created"}


def _verify_client_credentials(
    base_url: str,
    *,
    consumer_key: str,
    consumer_secret: str,
    verification_access_token: str,
) -> None:
    """Prove the stored secret by issuing, using, and revoking a token."""
    _verify_wrong_client_secret_is_rejected(
        base_url,
        consumer_key=consumer_key,
        consumer_secret=consumer_secret,
        verification_access_token=verification_access_token,
    )
    token_request = Request(
        f"{base_url.rstrip('/')}/services/oauth2/token",
        data=urlencode(
            {
                "grant_type": "client_credentials",
                "client_id": consumer_key,
                "client_secret": consumer_secret,
            }
        ).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urlopen(token_request, timeout=30) as response:
            token_payload = json.loads(response.read().decode())
    except HTTPError as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App client_credentials verification failed for ConsumerKey "
            f"{consumer_key!r}: http {exc.code}"
        ) from exc
    except (URLError, OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App client_credentials verification failed for ConsumerKey "
            f"{consumer_key!r}"
        ) from exc

    if not isinstance(token_payload, dict):
        raise ConnectedAppsSeedAdapterError(
            "Connected App client_credentials verification returned a malformed "
            f"response for ConsumerKey {consumer_key!r}"
        )
    access_token = str(token_payload.get("access_token") or "").strip()
    if not access_token:
        raise ConnectedAppsSeedAdapterError(
            "Connected App client_credentials verification returned no access_token "
            f"for ConsumerKey {consumer_key!r}"
        )

    protected_resource_url = (
        f"{base_url.rstrip('/')}/services/data/{API_VERSION}/sobjects"
    )
    protected_resource_request = Request(
        protected_resource_url,
        headers={"Authorization": f"Bearer {access_token}"},
        method="GET",
    )
    try:
        with open_verification_request(
            protected_resource_request,
            timeout=30,
        ) as response:
            response.read()
    except HTTPError as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App issued token protected-resource verification failed for "
            f"ConsumerKey {consumer_key!r}: http {exc.code}"
        ) from exc
    except (URLError, OSError) as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App issued token protected-resource verification failed for "
            f"ConsumerKey {consumer_key!r}"
        ) from exc

    revoke_request = Request(
        f"{base_url.rstrip('/')}/services/oauth2/revoke",
        data=urlencode({"token": access_token}).encode(),
        headers={
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/x-www-form-urlencoded",
        },
        method="POST",
    )
    try:
        with urlopen(revoke_request, timeout=30):
            pass
    except (HTTPError, URLError, OSError) as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App verification token revocation failed for ConsumerKey "
            f"{consumer_key!r}"
        ) from exc

    revoked_token_request = Request(
        protected_resource_url,
        headers={"Authorization": f"Bearer {access_token}"},
        method="GET",
    )
    try:
        with open_verification_request(
            revoked_token_request,
            timeout=30,
        ) as response:
            response.read()
    except HTTPError as exc:
        exc.read()
        if exc.code == 401:
            return
        raise ConnectedAppsSeedAdapterError(
            "Connected App revoked token protected-resource verification returned an "
            f"unexpected denial for ConsumerKey {consumer_key!r}: http {exc.code}"
        ) from exc
    except (URLError, OSError) as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App revoked token protected-resource verification failed for "
            f"ConsumerKey {consumer_key!r}"
        ) from exc
    else:
        raise ConnectedAppsSeedAdapterError(
            "Connected App verification token remains active after revocation for "
            f"ConsumerKey {consumer_key!r}"
        )


def _verify_wrong_client_secret_is_rejected(
    base_url: str,
    *,
    consumer_key: str,
    consumer_secret: str,
    verification_access_token: str,
) -> None:
    wrong_secret = hashlib.sha256(
        b"nexuscrm-connected-app-negative-secret\0" + consumer_secret.encode()
    ).hexdigest()
    if wrong_secret == consumer_secret:
        raise ConnectedAppsSeedAdapterError(
            "Connected App negative credential probe could not derive a distinct secret"
        )
    request = Request(
        f"{base_url.rstrip('/')}/services/oauth2/token",
        data=urlencode(
            {
                "grant_type": "client_credentials",
                "client_id": consumer_key,
                "client_secret": wrong_secret,
            }
        ).encode(),
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        method="POST",
    )
    try:
        with urlopen(request, timeout=30) as response:
            payload = json.loads(response.read().decode())
    except HTTPError as exc:
        try:
            protocol_error = json.loads(exc.read().decode())
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as decode_exc:
            raise ConnectedAppsSeedAdapterError(
                "Connected App wrong-secret verification returned a malformed denial for "
                f"ConsumerKey {consumer_key!r}"
            ) from decode_exc
        if (
            exc.code == 400
            and isinstance(protocol_error, dict)
            and protocol_error.get("error") == "invalid_client"
        ):
            return
        raise ConnectedAppsSeedAdapterError(
            "Connected App wrong-secret verification returned an unexpected denial for "
            f"ConsumerKey {consumer_key!r}: http {exc.code}"
        ) from exc
    except (URLError, OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise ConnectedAppsSeedAdapterError(
            "Connected App wrong-secret verification failed for ConsumerKey "
            f"{consumer_key!r}"
        ) from exc

    if not isinstance(payload, dict):
        raise ConnectedAppsSeedAdapterError(
            "Connected App accepted a wrong secret with a malformed response for ConsumerKey "
            f"{consumer_key!r}"
        )
    issued_token = str(
        payload.get("access_token") or payload.get("refresh_token") or ""
    ).strip()
    if issued_token:
        cleanup_request = Request(
            f"{base_url.rstrip('/')}/services/oauth2/revoke",
            data=urlencode({"token": issued_token}).encode(),
            headers={
                "Authorization": (
                    "Bearer "
                    + str(payload.get("access_token") or verification_access_token)
                ),
                "Content-Type": "application/x-www-form-urlencoded",
            },
            method="POST",
        )
        try:
            with urlopen(cleanup_request, timeout=30):
                pass
        except (HTTPError, URLError, OSError) as exc:
            raise ConnectedAppsSeedAdapterError(
                "Connected App wrong-secret probe issued a token and cleanup failed for "
                f"ConsumerKey {consumer_key!r}"
            ) from exc
    raise ConnectedAppsSeedAdapterError(
        "Connected App accepted a wrong client secret for ConsumerKey "
        f"{consumer_key!r}"
    )


def apply_connected_apps_via_api(
    base_url: str,
    access_token: str,
    *,
    seed_file: Path,
    environment: dict[str, str],
) -> dict[str, object]:
    spec = load_connected_apps(seed_file)
    seeded_users: list[dict[str, object]] = []
    applied_records: list[dict[str, object]] = []

    # Fail-closed before mutation: resolve every {{ env.* }} field ref
    # (e.g. ConsumerSecret) up front. Record refs pass through untouched and
    # are bound to the resolved User Id below.
    resolved_apps: list[dict[str, Any]] = []
    for app in spec["connected_apps"]:
        resolved_fields = {
            name: _resolve_env_ref(value, field=name, environment=environment)
            for name, value in app["fields"].items()
        }
        resolved_apps.append({**app, "fields": resolved_fields})

    # ConsumerSecret is non-readable, so an existing credential cannot be
    # rolled back after a bad rotation. Before any User or Connected App write,
    # prove that the runner secret already authenticates every existing app.
    # Existing-app PATCHes then omit ConsumerSecret and preserve its stored
    # hash. A missing app still receives the resolved secret on create.
    existing_app_ids: dict[str, str | None] = {}
    for app in resolved_apps:
        consumer_key = app["consumer_key"]
        existing_record_id = _lookup_connected_app(
            base_url,
            access_token,
            consumer_key=consumer_key,
        )
        existing_app_ids[consumer_key] = existing_record_id
        if existing_record_id is not None:
            _verify_client_credentials(
                base_url,
                consumer_key=consumer_key,
                consumer_secret=str(app["fields"]["ConsumerSecret"]),
                verification_access_token=access_token,
            )

    user_ids: dict[str, str] = {}
    for key, user in spec["users"].items():
        result = _get_or_create_user(
            base_url,
            access_token,
            username=user["username"],
            create_fields=user["create_fields"],
            existing_profile_name=user["existing_profile_name"],
        )
        user_ids[key] = result["id"]
        seeded_users.append(
            {
                "object": "User",
                "match": {"Username": user["username"]},
                "id": result["id"],
                "action": result["action"],
            }
        )

    for app in resolved_apps:
        fields = dict(app["fields"])
        fields["RunAsUserId"] = user_ids[app["run_as_user_key"]]
        result = _upsert_connected_app(
            base_url,
            access_token,
            consumer_key=app["consumer_key"],
            existing_record_id=existing_app_ids[app["consumer_key"]],
            fields=fields,
        )
        applied_records.append(
            {
                "object": "_System_ConnectedApp",
                "match": {"ConsumerKey": app["consumer_key"]},
                "id": result["id"],
                "action": result["action"],
                "run_as_user_id": fields["RunAsUserId"],
            }
        )

    return {"applied_records": applied_records, "seeded_users": seeded_users}


def verify_connected_apps(envv: dict[str, str], *, seed_file: Path) -> None:
    base_url = f"http://127.0.0.1:{envv['BACKEND_PORT']}"
    access_token = deploy_offline_env.request_exp_local_access_token(envv)
    spec = load_connected_apps(seed_file)

    user_ids: dict[str, str] = {}
    for key, user in spec["users"].items():
        username = user["username"]
        runtime_fields = _resolve_runtime_user_fields(
            base_url,
            access_token,
            seed_fields=user["create_fields"],
        )
        expected_profile_name = _profile_display_name(
            user["create_fields"].get("ProfileId")
        )
        records = _query_users_by_username(
            base_url,
            access_token,
            username=username,
        )
        if len(records) != 1:
            raise ConnectedAppsSeedAdapterError(
                "Connected App run-as User verify requires exactly one User for "
                f"{username!r}; found {len(records)}"
            )
        user_id = _require_user_matches_seed(
            records[0],
            username=username,
            expected_fields=runtime_fields,
            expected_profile_name=expected_profile_name,
            existing_profile_name=user["existing_profile_name"],
        )
        user_ids[key] = user_id

    verified = 0
    for app in spec["connected_apps"]:
        consumer_key = app["consumer_key"]
        expected = app["fields"]
        consumer_secret = str(
            _resolve_env_ref(
                expected["ConsumerSecret"],
                field="ConsumerSecret",
                environment=envv,
            )
        )
        expected_run_as = user_ids[app["run_as_user_key"]]
        # Do not select or compare ConsumerSecret: the runtime persists only its
        # hash. The token exchange below is the authoritative, non-readback proof
        # that the configured secret is usable.
        records = deploy_offline_env.exp_soql_records(
            base_url,
            access_token,
            (
                "SELECT Id, Name, ConsumerKey, IsActive, Scopes, "
                "PermittedUsers, RunAsUserId FROM _System_ConnectedApp "
                "WHERE ConsumerKey = "
                f"'{deploy_offline_env.soql_string_literal(consumer_key)}'"
            ),
        )
        if len(records) != 1:
            raise ConnectedAppsSeedAdapterError(
                "Connected App verify requires exactly one row for ConsumerKey "
                f"{consumer_key!r}; found {len(records)}"
            )
        record = records[0]
        actual_run_as = str(record.get("RunAsUserId") or "").strip()
        if actual_run_as != expected_run_as:
            raise ConnectedAppsSeedAdapterError(
                "Connected App RunAsUserId mismatch for "
                f"{consumer_key!r}: expected {expected_run_as}, found {actual_run_as}"
            )
        for field in ("Name", "Scopes", "PermittedUsers"):
            actual = str(record.get(field) or "").strip()
            if actual != str(expected[field]).strip():
                raise ConnectedAppsSeedAdapterError(
                    f"Connected App {field} mismatch for {consumer_key!r}: "
                    f"expected {expected[field]!r}, found {actual!r}"
                )
        if str(record.get("IsActive")).strip().lower() not in {"1", "true"}:
            raise ConnectedAppsSeedAdapterError(
                f"Connected App {consumer_key!r} is not active: {record.get('IsActive')!r}"
            )
        _verify_client_credentials(
            base_url,
            consumer_key=consumer_key,
            consumer_secret=consumer_secret,
            verification_access_token=access_token,
        )
        verified += 1
    deploy_offline_env.log(f"verified PingCAP connected apps count={verified}")


def _write_summary(path: Path, payload: dict[str, Any]) -> None:
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )


def run_adapter(args: argparse.Namespace) -> tuple[dict[str, Any], Path]:
    started = time.perf_counter()
    source_root = Path(args.source_root).expanduser().resolve(strict=True)
    if args.module_id != MODULE_ID:
        raise ConnectedAppsSeedAdapterError(
            "module id is not the Connected App Seed module"
        )
    _selection(args.selection_json)
    seed_file = _seed_file(source_root, args.data_source)
    spec = load_connected_apps(seed_file)
    if len(spec["connected_apps"]) != PLANNED_COUNTS["connected_apps"] or len(
        spec["users"]
    ) != PLANNED_COUNTS["seed_users"]:
        raise ConnectedAppsSeedAdapterError(
            "Connected App Seed count does not match the contract"
        )

    output_dir = Path(args.output_dir).expanduser().resolve()
    output_dir.mkdir(parents=True, exist_ok=False)
    applied_counts = {name: 0 for name in PLANNED_COUNTS}
    verification_status = "planned"
    if not args.dry_run:
        environment = dict(os.environ)
        backend_port = environment.get("BACKEND_PORT", "").strip()
        if not backend_port:
            raise ConnectedAppsSeedAdapterError("BACKEND_PORT is required in apply mode")
        base_url = f"http://127.0.0.1:{backend_port}"
        access_token = deploy_offline_env.request_exp_local_access_token(environment)
        payload = apply_connected_apps_via_api(
            base_url,
            access_token,
            seed_file=seed_file,
            environment=environment,
        )
        applied_counts = {
            "connected_apps": len(payload.get("applied_records", []) or []),
            "seed_users": len(payload.get("seeded_users", []) or []),
        }
        verify_connected_apps(environment, seed_file=seed_file)
        verification_status = "passed"

    summary = {
        "schema_version": 1,
        "format": SUMMARY_FORMAT,
        "module_id": MODULE_ID,
        "dry_run": args.dry_run,
        "apply_engine": "rest_api",
        "planned_counts": dict(PLANNED_COUNTS),
        "applied_counts": applied_counts,
        "coverage_complete": applied_counts == PLANNED_COUNTS if not args.dry_run else True,
        "verification_status": verification_status,
        "duration_ms": round((time.perf_counter() - started) * 1000),
    }
    summary_path = output_dir / SUMMARY_FILENAME
    _write_summary(summary_path, summary)
    print(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True))
    return summary, summary_path


def main(argv: list[str] | None = None) -> int:
    args = parse_args(argv)
    try:
        run_adapter(args)
    except (OSError, deploy_offline_env.DeployError, ConnectedAppsSeedAdapterError) as exc:
        print(f"PingCAP Connected App Seed adapter failed: {exc}", file=sys.stderr)
        return 2
    return 0


if __name__ == "__main__":
    raise SystemExit(main())