import json
import subprocess
import sys
import time

import boto3
import requests

AUTH0_TOKEN_URL = "https://tidb-soc2.us.auth0.com/oauth/token"
AUTH0_AUDIENCE = "https://tidb-soc2.us.auth0.com/api/v2/"


def get_auth0_token(auth0_secret_string: str) -> str:
    """Fetch an Auth0 bearer token using credentials in the secret value JSON string."""
    secret = json.loads(auth0_secret_string)
    client_id = secret["Auth0ClientID"]
    client_secret = secret["Auth0ClientSecret"]

    resp = requests.post(
        AUTH0_TOKEN_URL,
        data={
            "grant_type": "client_credentials",
            "client_id": client_id,
            "client_secret": client_secret,
            "audience": AUTH0_AUDIENCE,
        },
    )
    resp.raise_for_status()
    return f"Bearer {resp.json()['access_token']}"


def parse_account_id_from_role_arn(role_arn: str) -> str:
    parts = role_arn.split(":")
    if len(parts) < 6 or len(parts[4]) != 12 or not parts[4].isdigit():
        raise ValueError(f"invalid vector role arn: {role_arn}")
    if not parts[5].startswith("role/"):
        raise ValueError(f"invalid vector role arn: {role_arn}")
    return parts[4]


def load_auth0_secret_string_aws(secret_id: str, profile_name: str, region: str) -> str:
    resp = (
        boto3.Session(profile_name=profile_name)
        .client(
            "secretsmanager",
            region_name=region,
        )
        .get_secret_value(SecretId=secret_id)
    )
    secret_string = resp.get("SecretString")
    if not secret_string:
        raise RuntimeError(f"secret {secret_id} has empty SecretString")
    return secret_string


def load_auth0_secret_string_alicloud(secret_id: str, profile_name: str, region: str) -> str:
    result = subprocess.run(
        ["aliyun", "kms", "GetSecretValue", "--SecretName", secret_id, "--region", region, "--profile", profile_name],
        capture_output=True,
        text=True,
        check=True,
    )
    data = json.loads(result.stdout)
    secret_string = data.get("SecretData")
    if not secret_string:
        raise RuntimeError(f"secret {secret_id} has empty SecretData")
    return secret_string


def load_auth0_secret_string_azure(
    secret_id: str, key_vault_name: str, subscription_id: str | None = None
) -> str:
    command = [
        "az",
        "keyvault",
        "secret",
        "show",
        "--vault-name",
        key_vault_name,
        "--name",
        secret_id,
        "--query",
        "value",
        "-o",
        "tsv",
    ]
    if subscription_id:
        command.extend(["--subscription", subscription_id])

    result = subprocess.run(
        command,
        capture_output=True,
        text=True,
        check=True,
    )
    secret_string = result.stdout.strip()
    if not secret_string:
        raise RuntimeError(f"secret {secret_id} has empty value in key vault {key_vault_name}")
    return secret_string


class O11yClient:
    def __init__(self, endpoint: str, token: str, tenant_id: str, biz_type: str):
        self.endpoint = endpoint
        self.token = token
        self.tenant_id = tenant_id
        self.biz_type = biz_type

    @property
    def _headers(self) -> dict:
        return {
            "Authorization": self.token,
            "Content-Type": "application/json",
            "X-Biz-Type": self.biz_type,
            "X-Org-Id": self.tenant_id,
        }

    def create_cluster(self, body: dict) -> dict:
        resp = requests.post(
            f"http://{self.endpoint}/api/v1/clusters/",
            headers=self._headers,
            json=body,
        )
        resp.raise_for_status()
        return resp.json()

    def get_cluster(self, cluster_id: str, vendor: str, region: str) -> dict:
        resp = requests.get(
            f"http://{self.endpoint}/api/v1/clusters/{cluster_id}",
            headers=self._headers,
            params={"vendor": vendor, "region": region},
        )
        resp.raise_for_status()
        return resp.json()

    def delete_cluster(self, cluster_id: str, vendor: str, region: str) -> dict:
        resp = requests.delete(
            f"http://{self.endpoint}/api/v1/clusters/{cluster_id}",
            headers=self._headers,
            params={"vendor": vendor, "region": region},
        )
        resp.raise_for_status()
        return resp.json()

    def list_clusters(self, params: dict) -> list[dict]:
        resp = requests.get(
            f"http://{self.endpoint}/api/v1/clusters",
            headers=self._headers,
            params=params,
        )
        resp.raise_for_status()
        return resp.json().get("data", {}).get("clusters", [])

    def wait_for_active(
        self,
        cluster_id: str,
        vendor: str,
        region: str,
        timeout: int = 300,
        interval: int = 10,
    ) -> dict:
        """Poll until the cluster status is ACTIVE, then return the cluster data dict."""
        start = time.time()
        while time.time() - start < timeout:
            cluster_data = self.get_cluster(cluster_id, vendor, region).get("data", {})
            status = cluster_data.get("status")
            print(f"o11y cluster status: {status}", file=sys.stderr)
            if status == "ACTIVE":
                print("o11y cluster is ready!", file=sys.stderr)
                return cluster_data
            time.sleep(interval)
        raise TimeoutError(
            f"o11y cluster {cluster_id} did not become ACTIVE within {timeout}s"
        )