"""Dev/staging environment cluster cleanup rules.
By default, clusters are deleted after 8 hours past creation times.
This can be overridden by specifying "keep-Nh" (e.g., keep-12h) in the cluster name.
Malformed values like "keep--1h" are normalized and treated as "keep-1h".

Uses grpc-requests (server reflection) to avoid maintaining generated pb stubs.
All three cluster types (dedicated=1, serverless=2, nextgen=3) are paginated
exhaustively so that large projects don't crowd out the results.

Project 3200705 (org 2130684) contains ~300k long-lived keep-9999h clusters
and is explicitly excluded at the API level via exclude_project_ids to avoid
fetching its clusters at all.
"""

import datetime
import logging
import os
import re
import typing

from grpc_requests import Client
import grpc
import requests

MGMT_SVC_ADDR = os.getenv(
    "MGMT_SVC_ADDR", "tidb-mgmt-service.tidb-management-service:10001"
)
KEEP_HOUR = int(os.getenv("keep_hour", "8"))
DRY_RUN = os.getenv("dryrun", "false").lower() in ("true", "1", "t")
FEISHU_WEBHOOK_URL = (
    "https://open.feishu.cn/open-apis/bot/v2/hook/bd35d09e-2074-4811-8422-ba21b6b948ac"
)

# Project IDs that should never be touched (contain permanent long-lived clusters).
EXCLUDE_PROJECT_IDS = [3200705]

# None: unknown (default, try with exclude_project_ids first)
# True: supported by current endpoint
# False: not supported by current endpoint, fallback without this field
SUPPORTS_EXCLUDE_PROJECT_IDS: typing.Optional[bool] = None

CLUSTER_REGEX_PATTERN = r"keep-(\d+)h"
CLUSTER_TYPO_DOUBLE_MINUS_PATTERN = r"keep--(\d+)h"

FSN_CLUSTER_SERVICE = "tidb_mgmt_service.v1.ClusterService"
MN_LIST_CLUSTERS = "ListClusters"
MN_DELETE_CLUSTER = "DeleteCluster"

# cluster_type values (matches DeployType enum in proto)
DEPLOY_TYPE_DEDICATED = 1
DEPLOY_TYPE_SERVERLESS = 2
DEPLOY_TYPE_NEXTGEN = 3
DEPLOY_TYPE_NAMES = {
    DEPLOY_TYPE_DEDICATED: "DEDICATED",
    DEPLOY_TYPE_SERVERLESS: "SERVERLESS",
    DEPLOY_TYPE_NEXTGEN: "NEXTGEN",
}
DEFAULT_DEPLOY_TYPES = (
    DEPLOY_TYPE_DEDICATED,
    DEPLOY_TYPE_SERVERLESS,
    DEPLOY_TYPE_NEXTGEN,
)
DEPLOY_TYPE_ALIASES = {
    "1": DEPLOY_TYPE_DEDICATED,
    "DEDICATED": DEPLOY_TYPE_DEDICATED,
    "DEPLOY_TYPE_DEDICATED": DEPLOY_TYPE_DEDICATED,
    "2": DEPLOY_TYPE_SERVERLESS,
    "SERVERLESS": DEPLOY_TYPE_SERVERLESS,
    "SHARED": DEPLOY_TYPE_SERVERLESS,
    "DEPLOY_TYPE_SERVERLESS": DEPLOY_TYPE_SERVERLESS,
    "DEPLOY_TYPE_SHARED": DEPLOY_TYPE_SERVERLESS,
    "3": DEPLOY_TYPE_NEXTGEN,
    "NEXTGEN": DEPLOY_TYPE_NEXTGEN,
    "NEXT_GEN": DEPLOY_TYPE_NEXTGEN,
    "HYBRID": DEPLOY_TYPE_NEXTGEN,
    "DEPLOY_TYPE_NEXTGEN": DEPLOY_TYPE_NEXTGEN,
    "DEPLOY_TYPE_HYBRID": DEPLOY_TYPE_NEXTGEN,
}

PAGE_SIZE = 1000  # server maxPageSize is 1000
CUSTOM_METADATA = {}

LOG_FORMAT = "%(asctime)s - %(levelname)s - %(message)s"
DATE_FORMAT = "%m/%d/%Y %H:%M:%S %p"
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT, datefmt=DATE_FORMAT)


def _parse_deploy_types(value: str) -> typing.List[int]:
    """Parses DEPLOY_TYPE into the cluster_type values used by ListClusters."""
    if not value or value.strip().upper() in ("ALL", "*"):
        return list(DEFAULT_DEPLOY_TYPES)

    deploy_types: typing.List[int] = []
    for raw_token in re.split(r"[,\s]+", value):
        token = raw_token.strip()
        if not token:
            continue

        normalized = token.upper().replace("-", "_")
        deploy_type = DEPLOY_TYPE_ALIASES.get(normalized)
        if deploy_type is None:
            valid_values = ", ".join(sorted(DEPLOY_TYPE_ALIASES))
            raise ValueError(
                f"Unsupported DEPLOY_TYPE value {token!r}. Valid values: {valid_values}, ALL"
            )
        if deploy_type not in deploy_types:
            deploy_types.append(deploy_type)

    return deploy_types or list(DEFAULT_DEPLOY_TYPES)


def _format_deploy_types(deploy_types: typing.Sequence[int]) -> str:
    return ", ".join(DEPLOY_TYPE_NAMES.get(t, str(t)) for t in deploy_types)


def _get_client(addr: str) -> Client:
    """Initializes and returns a grpc-requests Client using server reflection."""
    logging.debug(f"Establishing gRPC client to {addr}")
    return Client.get_by_endpoint(addr)


def _grpc_request(
    client: Client,
    service_name: str,
    method_name: str,
    request_data: dict,
    metadata: dict,
) -> typing.Any:
    """Performs a dynamic gRPC request and handles common errors."""
    metadata_list = [(k.lower(), str(v)) for k, v in metadata.items()]
    full_method_name = f"{service_name}/{method_name}"
    logging.debug(f"Request: {full_method_name}, Data: {request_data}")

    try:
        response = client.request(
            service_name,
            method_name,
            request_data,
            metadata=metadata_list,
            timeout=30,
        )
        logging.debug(f"Response: {response}")
        return response

    except grpc.RpcError as e:
        error_details = f"Status Code: {e.code().name}, Details: {e.details()}"
        logging.error(f"gRPC call failed for {full_method_name}: {error_details}")
        raise ConnectionError(f"gRPC RPC Error: {error_details}") from e
    except Exception as e:
        logging.error(f"Unexpected error during gRPC call to {full_method_name}: {e}")
        raise RuntimeError(f"Unexpected gRPC Error: {e}") from e


def _list_all_clusters(client: Client, cluster_type: int) -> typing.List[dict]:
    """
    Fetches all clusters of the given type, paginating through all pages.
    Passes exclude_project_ids to skip the large permanent-cluster project.
    """
    all_clusters: typing.List[dict] = []
    page = 1
    type_name = DEPLOY_TYPE_NAMES.get(cluster_type, str(cluster_type))

    global SUPPORTS_EXCLUDE_PROJECT_IDS

    while True:
        include_exclude_field = SUPPORTS_EXCLUDE_PROJECT_IDS is not False
        filter_data = {
            "view": 1,  # VIEW_WITH_USER_INFO
            "deleted": False,
            "cluster_type": cluster_type,
        }
        if include_exclude_field:
            filter_data["exclude_project_ids"] = EXCLUDE_PROJECT_IDS

        request_data = {
            "filter": filter_data,
            "page": page,
            "page_size": PAGE_SIZE,
        }

        logging.info(f"[{type_name}] Fetching page {page} (page_size={PAGE_SIZE})...")
        try:
            response = _grpc_request(
                client,
                FSN_CLUSTER_SERVICE,
                MN_LIST_CLUSTERS,
                request_data,
                CUSTOM_METADATA,
            )
            if include_exclude_field and SUPPORTS_EXCLUDE_PROJECT_IDS is None:
                SUPPORTS_EXCLUDE_PROJECT_IDS = True
        except RuntimeError as e:
            err_str = str(e)
            if (
                include_exclude_field
                and 'has no field named "exclude_project_ids"' in err_str
            ):
                SUPPORTS_EXCLUDE_PROJECT_IDS = False
                logging.warning(
                    "Current endpoint does not support exclude_project_ids; "
                    "falling back without this field."
                )
                request_data["filter"].pop("exclude_project_ids", None)
                response = _grpc_request(
                    client,
                    FSN_CLUSTER_SERVICE,
                    MN_LIST_CLUSTERS,
                    request_data,
                    CUSTOM_METADATA,
                )
            else:
                raise

        clusters = response.get("clusters", [])
        all_clusters.extend(clusters)
        logging.info(
            f"[{type_name}] Page {page}: got {len(clusters)} clusters "
            f"(total so far: {len(all_clusters)})"
        )

        # Stop when we receive fewer results than a full page — no more pages.
        if len(clusters) < PAGE_SIZE:
            break
        page += 1

    return all_clusters


def _get_keep_hour(cluster_name: str) -> int:
    """
    Returns the retention hours for a cluster.
    Reads 'keep-Nh' from the cluster name; negative means keep forever (returns very large int).
    """
    typo_match = re.search(CLUSTER_TYPO_DOUBLE_MINUS_PATTERN, cluster_name)
    if typo_match:
        h = int(typo_match.group(1))
        logging.warning(
            f"Cluster name '{cluster_name}' contains malformed keep value; "
            f"normalizing keep--{h}h to keep-{h}h."
        )
        return h

    match = re.search(CLUSTER_REGEX_PATTERN, cluster_name)
    if match:
        return int(match.group(1))

    return KEEP_HOUR


def _parse_create_time(cluster: dict) -> typing.Optional[datetime.datetime]:
    """
    Parses create_time from a cluster dict.
    The field may be a Timestamp message dict {'seconds': N, 'nanos': N}
    or an ISO-8601 string.
    """
    ct = cluster.get("create_time")
    if ct is None:
        return None

    if isinstance(ct, dict):
        seconds = ct.get("seconds", 0)
        nanos = ct.get("nanos", 0)
        return datetime.datetime.fromtimestamp(
            seconds + nanos / 1e9, tz=datetime.timezone.utc
        )

    if isinstance(ct, str):
        try:
            return datetime.datetime.fromisoformat(ct.replace("Z", "+00:00"))
        except ValueError:
            logging.warning(f"Cannot parse create_time string: {ct!r}")
            return None

    return None


def process_cluster_cleanup(
    dry_run: bool = False,
    deploy_types: typing.Optional[typing.Sequence[int]] = None,
):
    """
    Lists live clusters for the selected DEPLOY_TYPE values, checks expiry,
    and deletes expired ones.
    """
    if deploy_types is None:
        deploy_types = _parse_deploy_types(os.getenv("DEPLOY_TYPE", ""))
    deploy_types = list(deploy_types)

    client = _get_client(MGMT_SVC_ADDR)
    deletion_failures: typing.List[str] = []
    skipped_forbidden: typing.List[str] = []
    deleted_clusters: typing.List[str] = []

    action_type = "DRY-RUN" if dry_run else "DELETION"
    logging.info(f"--- Starting Cluster Cleanup ({action_type}) ---")
    logging.info(
        f"Default retention: {KEEP_HOUR}h | Excluded project IDs: {EXCLUDE_PROJECT_IDS}"
    )
    logging.info(f"Target deploy types: {_format_deploy_types(deploy_types)}")

    # 1. Collect clusters from the selected types with full pagination.
    all_clusters: typing.List[dict] = []
    try:
        for cluster_type in deploy_types:
            clusters = _list_all_clusters(client, cluster_type)
            all_clusters.extend(clusters)
    except Exception as e:
        logging.critical(f"Failed to list clusters: {e}. Cannot proceed.")
        try:
            client.close()
        except AttributeError:
            pass
        raise RuntimeError("Cleanup terminated: cluster listing failed.") from e

    logging.info(f"Total clusters fetched: {len(all_clusters)}")

    # 2. Determine which clusters are eligible for deletion.
    now = datetime.datetime.now(tz=datetime.timezone.utc)
    pending_cleanup: typing.List[dict] = []

    for cluster in all_clusters:
        cluster_id = str(cluster.get("cluster_id", "N/A"))
        cluster_name = cluster.get("cluster_name", "")
        deploy_type = cluster.get("deploy_type", 0)
        deploy_type_name = DEPLOY_TYPE_NAMES.get(deploy_type, str(deploy_type))
        log_prefix = f"[{cluster_name}][{cluster_id}][{deploy_type_name}]"

        create_time = _parse_create_time(cluster)
        if create_time is None:
            logging.warning(
                f"{log_prefix} Skipping: missing or unparseable create_time."
            )
            continue

        keep_hour = _get_keep_hour(cluster_name)
        age = now - create_time

        if age > datetime.timedelta(hours=keep_hour):
            logging.warning(
                f"{log_prefix} ELIGIBLE FOR DELETION. keep={keep_hour}h, age={age}"
            )
            pending_cleanup.append(cluster)
        else:
            logging.info(f"{log_prefix} Within retention. keep={keep_hour}h, age={age}")

    logging.info(f"Clusters eligible for deletion: {len(pending_cleanup)}")

    if not pending_cleanup:
        logging.info("No clusters to delete. Cleanup finished.")
        try:
            client.close()
        except AttributeError:
            pass
        return

    if dry_run:
        for cluster in pending_cleanup:
            cluster_id = str(cluster.get("cluster_id", "N/A"))
            cluster_name = cluster.get("cluster_name", "")
            deploy_type = cluster.get("deploy_type", 0)
            logging.info(
                f"[DRY-RUN] Would delete: {cluster_name}[{cluster_id}] deploy_type={deploy_type}"
            )
        try:
            client.close()
        except AttributeError:
            pass
        return

    # 3. Delete eligible clusters.
    deleted_count = 0
    for cluster in pending_cleanup:
        cluster_id = str(cluster.get("cluster_id", "N/A"))
        cluster_name = cluster.get("cluster_name", "")
        deploy_type = cluster.get("deploy_type", 0)
        log_prefix = f"[{cluster_name}][{cluster_id}]"

        delete_request = {
            "cluster_id": cluster_id,
            "deploy_type": deploy_type,
        }

        try:
            logging.info(f"{log_prefix} Deleting...")
            _grpc_request(
                client,
                FSN_CLUSTER_SERVICE,
                MN_DELETE_CLUSTER,
                delete_request,
                CUSTOM_METADATA,
            )
            logging.info(f"{log_prefix} Deleted successfully.")
            deleted_count += 1
            deleted_clusters.append(
                f"{cluster_name}[{cluster_id}][{DEPLOY_TYPE_NAMES.get(deploy_type, str(deploy_type))}]"
            )
        except Exception as e:
            error_str = str(e)
            error_msg = f"Deletion failed for cluster {cluster_id}: {error_str}"

            # Some clusters are expected to be non-deletable by this recycler.
            # Do not fail the whole job for these known forbidden states.
            if (
                "Forbidden." in error_str
                and "cannot execute this operation" in error_str
            ):
                logging.warning(f"{log_prefix} Skip non-deletable cluster: {error_str}")
                skipped_forbidden.append(error_msg)
                continue

            logging.error(f"{log_prefix} {error_msg}")
            deletion_failures.append(error_msg)

    logging.info(
        f"Cleanup finished. Deleted {deleted_count} / {len(pending_cleanup)} eligible clusters."
    )
    if deleted_clusters:
        logging.info(
            "--- Successfully deleted clusters ---\n%s", "\n".join(deleted_clusters)
        )
    else:
        logging.info("No clusters were deleted in this run.")

    if skipped_forbidden:
        logging.warning(
            f"Skipped {len(skipped_forbidden)} non-deletable clusters due to forbidden states."
        )

    try:
        client.close()
    except AttributeError:
        pass

    if deletion_failures:
        summary = "\n".join(deletion_failures)
        logging.error(
            f"--- Deletion Failures ({len(deletion_failures)}) ---\n{summary}"
        )
        send_feishu_simple_notification(
            _build_failure_notification_message(deletion_failures)
        )
        raise RuntimeError(
            f"Cleanup finished with {len(deletion_failures)} deletion failure(s). See logs."
        )


def _build_failure_notification_message(deletion_failures: typing.Sequence[str]) -> str:
    header = f"cluster clean failed: {len(deletion_failures)} deletion(s) failed."
    if not deletion_failures:
        return header

    details = "\n".join(f"- {failure}" for failure in deletion_failures)
    return f"{header}\nFailed clusters and reasons:\n{details}"


def send_feishu_simple_notification(message: str):
    if not FEISHU_WEBHOOK_URL:
        logging.warning("FEISHU_WEBHOOK_URL is not set. Skipping Feishu notification.")
        return

    payload = {
        "msg_type": "text",
        "content": {
            "text": message,
        },
    }

    try:
        logging.info("Attempting to send simple Feishu notification...")
        response = requests.post(FEISHU_WEBHOOK_URL, json=payload, timeout=5)
        response.raise_for_status()
        logging.info("Simple Feishu notification sent successfully.")
    except requests.exceptions.RequestException as e:
        logging.error(f"Failed to send simple Feishu notification: {e}")


def process():
    """Entry point: reads DRY_RUN env var and runs cleanup."""
    process_cluster_cleanup(dry_run=DRY_RUN)


if __name__ == "__main__":
    process()