#!/usr/bin/env python3
"""
Detect long-stuck DM and CDC jobs from Grafana-backed SQL queries.

This cronjob persists local state for dedupe and sends a Feishu alert card only
when new abnormal items are detected.
"""

from __future__ import annotations

import argparse
import datetime as dt
import json
import os
from pathlib import Path
from typing import Any

import requests


GRAFANA_URL = "https://grafana.oauth.prod.aws.tidbcloud.com/api/ds/query"
GRAFANA_DATASOURCE_ID = 37
DEFAULT_STATE_FILE = "/tmp/cdc-dm-import-abnormal-check-state.json"
DM_CREATING_THRESHOLD_MINUTES = 30
CDC_DELETING_THRESHOLD_HOURS = 24
DEFAULT_NOTIFY_TARGET = os.environ.get("ONCALL_NOTIFY_CHAT_ID") or os.environ.get("ONCALL_METRICS_CHAT_ID") or "oc_2c586abb5204eca2ddf6ddf5de338bf6"
MAX_CARD_ITEMS = 8
GRAFANA_DM_URL = "https://grafana.oauth.prod.aws.tidbcloud.com/d/9sVoewCIz/global-control-user-metrics"
GRAFANA_CDC_URL = "https://grafana.oauth.prod.aws.tidbcloud.com/d/9sVoewCIz/global-control-user-metrics"
CHINA_TZ = dt.timezone(dt.timedelta(hours=8))
INTERNAL_TENANTS = (
    "1372813089194211283",
    "1372813089187791289",
    "1372813089189381281",
    "1372813089208161328",
    "1372813089188691290",
    "1372813089209308952",
)


def load_env_from_bashrc() -> None:
    """Load exported environment variables from ~/.bashrc."""
    bashrc = os.path.expanduser("~/.bashrc")
    if not os.path.exists(bashrc):
        return

    with open(bashrc, encoding="utf-8") as handle:
        for line in handle:
            line = line.strip()
            if not line.startswith("export ") or "=" not in line:
                continue
            rest = line.replace("export ", "", 1)
            key = rest.split("=", 1)[0].strip()
            value = rest.split("=", 1)[1].strip().strip('"').strip("'")
            os.environ.setdefault(key, os.path.expanduser(os.path.expandvars(value)))


load_env_from_bashrc()

GRAFANA_TOKEN = os.environ.get("DATASERVICE_GRAFANA_TOKEN", "")
FEISHU_APP_ID = os.environ.get("CRONJOB_TEAM_WORKFLOW_APP_ID", "")
FEISHU_APP_SECRET = os.environ.get("CRONJOB_TEAM_WORKFLOW_APP_SECRET", "")


def normalize_text(value: Any, default: str = "") -> str:
    if value is None:
        return default
    text = str(value).strip()
    return text or default


def normalize_json_error(value: Any) -> str:
    text = normalize_text(value)
    if not text:
        return ""
    try:
        parsed = json.loads(text)
    except Exception:
        return text
    if isinstance(parsed, dict):
        error_msg = normalize_text(parsed.get("errorMsg"))
        if error_msg:
            return error_msg
    return text


def is_true(value: Any) -> bool:
    return str(value or "").strip().lower() in {"1", "true", "yes", "on"}


def shorten(text: str, limit: int = 200) -> str:
    flattened = " ".join(normalize_text(text).split())
    if len(flattened) <= limit:
        return flattened
    return flattened[: limit - 3].rstrip() + "..."


def get_feishu_token() -> str | None:
    if not FEISHU_APP_ID or not FEISHU_APP_SECRET:
        print("Feishu app credentials are not set, skip notification")
        return None

    response = requests.post(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        headers={"Content-Type": "application/json"},
        json={"app_id": FEISHU_APP_ID, "app_secret": FEISHU_APP_SECRET},
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(f"failed to get Feishu token: {data.get('msg')}")
    return data.get("tenant_access_token")


def parse_feishu_target(target: str) -> tuple[str, str]:
    value = normalize_text(target)
    if not value:
        raise ValueError("empty Feishu target")
    if value.startswith("user:"):
        return "open_id", value.split(":", 1)[1]
    if value.startswith("chat:"):
        return "chat_id", value.split(":", 1)[1]
    if value.startswith("ou_"):
        return "open_id", value
    return "chat_id", value


def query_grafana_rows(sql: str, lookback_minutes: int) -> list[list[Any]]:
    """Query Grafana SQL datasource and return row-oriented results."""
    if not GRAFANA_TOKEN:
        raise RuntimeError("DATASERVICE_GRAFANA_TOKEN is not set")

    payload = {
        "queries": [
            {
                "refId": "A",
                "datasourceId": GRAFANA_DATASOURCE_ID,
                "rawSql": sql,
                "format": "table",
            }
        ],
        "from": f"now-{lookback_minutes}m",
        "to": "now",
    }
    response = requests.post(
        GRAFANA_URL,
        headers={"Authorization": f"Bearer {GRAFANA_TOKEN}", "Content-Type": "application/json"},
        json=payload,
        timeout=120,
    )
    response.raise_for_status()
    data = response.json()
    frames = data.get("results", {}).get("A", {}).get("frames", [])
    if not frames:
        return []

    values = frames[0].get("data", {}).get("values", [])
    if not values:
        return []

    row_count = len(values[0]) if values else 0
    return [[values[col][row] if row < len(values[col]) else None for col in range(len(values))] for row in range(row_count)]

def query_dm_abnormal_jobs() -> list[dict[str, Any]]:
    sql = f"""
    SELECT
        CONCAT(a.id, '') AS job_id,
        CONCAT(a.cluster_id, '') AS cluster_id,
        CONCAT(a.project_id, '') AS project_id,
        CONCAT(a.tenant_id, '') AS tenant_id,
        a.phase,
        a.created_at,
        a.updated_at,
        a.error_message
    FROM df_jobs a
    WHERE a.service_code = 'dm'
      AND a.tenant_id NOT IN ({','.join(INTERNAL_TENANTS)})
      AND a.phase = 'creating'
      AND a.created_at <= DATE_SUB(NOW(), INTERVAL {DM_CREATING_THRESHOLD_MINUTES} MINUTE)
    ORDER BY a.created_at ASC
    LIMIT 1000
    """
    rows = query_grafana_rows(sql, DM_CREATING_THRESHOLD_MINUTES * 4)

    anomalies = []
    for row in rows:
        summary = normalize_json_error(row[7] if len(row) > 7 else "")
        if not summary:
            summary = f"DM job stayed in creating for over {DM_CREATING_THRESHOLD_MINUTES} minutes"
        anomalies.append(
            {
                "component": "DM",
                "job_id": normalize_text(row[0], "unknown"),
                "cluster": normalize_text(row[1], "unknown"),
                "tenant_id": normalize_text(row[3]),
                "project_id": normalize_text(row[2]),
                "occurred_at": normalize_text(row[5]),
                "status": normalize_text(row[4], "creating"),
                "summary": shorten(summary),
                "detail_url": GRAFANA_DM_URL,
            }
        )

    return anomalies


def query_cdc_abnormal_jobs() -> list[dict[str, Any]]:
    sql = f"""
    SELECT
        CAST(id AS CHAR),
        CAST(cluster_id AS CHAR),
        CAST(project_id AS CHAR),
        CAST(tenant_id AS CHAR),
        operation_status,
        updated_at,
        created_at
    FROM biz_changefeeds
    WHERE operation_status = 'deleting'
      AND updated_at <= DATE_SUB(NOW(), INTERVAL {CDC_DELETING_THRESHOLD_HOURS} HOUR)
      AND tenant_id NOT IN ({','.join(INTERNAL_TENANTS)})
    ORDER BY updated_at DESC
    LIMIT 1000
    """
    rows = query_grafana_rows(sql, CDC_DELETING_THRESHOLD_HOURS * 60 * 3)

    anomalies = []
    for row in rows:
        anomalies.append(
            {
                "component": "CDC",
                "job_id": normalize_text(row[0], "unknown"),
                "cluster": normalize_text(row[1], "unknown"),
                "tenant_id": normalize_text(row[3]),
                "project_id": normalize_text(row[2]),
                "occurred_at": normalize_text(row[5]) or normalize_text(row[6]),
                "status": normalize_text(row[4], "deleting"),
                "summary": f"CDC changefeed stayed in deleting for over {CDC_DELETING_THRESHOLD_HOURS} hours",
                "detail_url": GRAFANA_CDC_URL,
            }
        )

    return anomalies


def build_anomaly_key(item: dict[str, Any]) -> str:
    return "|".join(
        [
            normalize_text(item.get("component"), "unknown"),
            normalize_text(item.get("job_id"), "unknown"),
            normalize_text(item.get("status"), "unknown"),
        ]
    )


def load_state(state_file: str) -> dict[str, Any]:
    path = Path(state_file)
    if not path.exists():
        return {"active": {}}
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return {"active": {}}


def save_state(state_file: str, state: dict[str, Any]) -> None:
    path = Path(state_file)
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(state, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")


def reconcile_state(
    anomalies: list[dict[str, Any]], state: dict[str, Any], observed_at: str
) -> tuple[list[dict[str, Any]], list[dict[str, Any]], dict[str, Any]]:
    previous_active = state.get("active", {})
    current_active: dict[str, Any] = {}
    new_anomalies: list[dict[str, Any]] = []

    for anomaly in anomalies:
        key = build_anomaly_key(anomaly)
        previous = previous_active.get(key)
        entry = {
            "first_seen_at": previous.get("first_seen_at", observed_at) if previous else observed_at,
            "last_seen_at": observed_at,
            "occurrences": int(previous.get("occurrences", 0)) + 1 if previous else 1,
            "anomaly": anomaly,
        }
        current_active[key] = entry
        if not previous:
            new_anomalies.append(anomaly)

    resolved_anomalies = [
        previous_active[key]["anomaly"] for key in previous_active.keys() - current_active.keys()
    ]
    return new_anomalies, resolved_anomalies, {"active": current_active}


def build_summary(
    anomalies: list[dict[str, Any]],
    new_anomalies: list[dict[str, Any]],
    resolved_anomalies: list[dict[str, Any]],
    observed_at: str,
) -> dict[str, Any]:
    counts = {"DM": 0, "CDC": 0}
    for anomaly in anomalies:
        counts[anomaly["component"]] = counts.get(anomaly["component"], 0) + 1

    return {
        "observed_at": observed_at,
        "counts": counts,
        "total_abnormal": len(anomalies),
        "new_abnormal": len(new_anomalies),
        "resolved_abnormal": len(resolved_anomalies),
        "should_notify": bool(new_anomalies),
        "anomalies": anomalies,
        "new_items": new_anomalies,
        "resolved_items": resolved_anomalies,
    }


def format_text_summary(summary: dict[str, Any]) -> str:
    lines = [
        "CDC/DM abnormal check",
        f"Observed at: {summary['observed_at']}",
        (
            "Counts: "
            f"CDC {summary['counts'].get('CDC', 0)}, "
            f"DM {summary['counts'].get('DM', 0)}"
        ),
        f"New abnormal: {summary['new_abnormal']}",
        f"Resolved abnormal: {summary['resolved_abnormal']}",
    ]
    if not summary["anomalies"]:
        lines.append("No abnormal jobs detected.")
        return "\n".join(lines)

    lines.append("Active abnormal jobs:")
    for item in summary["anomalies"]:
        lines.append(
            (
                f"- [{item['component']}] job={item['job_id']} status={item['status']} "
                f"cluster={item['cluster']} tenant={item['tenant_id'] or '-'} "
                f"project={item['project_id'] or '-'} summary={item['summary']}"
            )
        )
    return "\n".join(lines)


def format_timestamp(value: str) -> str:
    text = normalize_text(value, "-")
    if text == "-":
        return text
    if text.isdigit():
        try:
            timestamp = int(text)
            if timestamp > 10**12:
                parsed = dt.datetime.fromtimestamp(timestamp / 1000, tz=dt.UTC)
            else:
                parsed = dt.datetime.fromtimestamp(timestamp, tz=dt.UTC)
            return parsed.astimezone(CHINA_TZ).strftime("%Y-%m-%d %H:%M:%S UTC+8")
        except Exception:
            return text
    try:
        parsed = dt.datetime.fromisoformat(text.replace("Z", "+00:00"))
        if parsed.tzinfo is None:
            parsed = parsed.replace(tzinfo=dt.UTC)
        return parsed.astimezone(CHINA_TZ).strftime("%Y-%m-%d %H:%M:%S UTC+8")
    except Exception:
        pass
    return text


def build_notification_card(summary: dict[str, Any]) -> dict[str, Any]:
    active_items = summary["new_items"] or summary["anomalies"]
    item_blocks: list[dict[str, Any]] = []
    for index, item in enumerate(active_items[:MAX_CARD_ITEMS], 1):
        item_blocks.append(
            {
                "tag": "markdown",
                "content": (
                    "**{index}. [{component}] {job_id}**\n"
                    "Status: {status}\n"
                    "Cluster: {cluster}\n"
                    "Tenant: {tenant}\n"
                    "Project: {project}\n"
                    "Observed Time: {occurred_at}\n"
                    "Alert: {summary}"
                ).format(
                    index=index,
                    component=item["component"],
                    job_id=normalize_text(item["job_id"]),
                    status=normalize_text(item["status"]),
                    cluster=normalize_text(item["cluster"], "-"),
                    tenant=normalize_text(item["tenant_id"], "-"),
                    project=normalize_text(item["project_id"], "-"),
                    occurred_at=format_timestamp(item["occurred_at"]),
                    summary=shorten(normalize_text(item["summary"]), 160),
                ),
            }
        )
        item_blocks.append({"tag": "hr"})

    if item_blocks:
        item_blocks.pop()
    else:
        item_blocks.append({"tag": "markdown", "content": "No active abnormal jobs."})

    if len(active_items) > MAX_CARD_ITEMS:
        item_blocks.extend(
            [
                {"tag": "hr"},
                {
                    "tag": "note",
                    "elements": [
                        {
                            "tag": "plain_text",
                            "content": f"{len(active_items) - MAX_CARD_ITEMS} more items were omitted from this card.",
                        }
                    ],
                },
            ]
        )

    severity_text = "high" if active_items else "info"
    elements: list[dict[str, Any]] = [
        {
            "tag": "markdown",
            "content": (
                f"**Observed At**: {format_timestamp(summary['observed_at'])}\n"
                f"**New Alerts**: {summary['new_abnormal']}\n"
                f"**Resolved Since Last Run**: {summary['resolved_abnormal']}"
            ),
        },
        {
            "tag": "column_set",
            "flex_mode": "none",
            "columns": [
                {
                    "tag": "column",
                    "width": "weighted",
                    "weight": 1,
                    "elements": [{"tag": "markdown", "content": f"**DM Stuck**\n{summary['counts'].get('DM', 0)}"}],
                },
                {
                    "tag": "column",
                    "width": "weighted",
                    "weight": 1,
                    "elements": [{"tag": "markdown", "content": f"**CDC Stuck**\n{summary['counts'].get('CDC', 0)}"}],
                },
                {
                    "tag": "column",
                    "width": "weighted",
                    "weight": 1,
                    "elements": [{"tag": "markdown", "content": f"**Severity**\n{severity_text}"}],
                },
            ],
        },
        {
            "tag": "note",
            "elements": [
                {
                    "tag": "plain_text",
                    "content": (
                        f"Rules: DM phase creating > {DM_CREATING_THRESHOLD_MINUTES} minutes; "
                        f"CDC status deleting > {CDC_DELETING_THRESHOLD_HOURS} hours"
                    ),
                }
            ],
        },
        {"tag": "hr"},
        {"tag": "markdown", "content": "**Alert Items**"},
    ]
    elements.extend(item_blocks)
    elements.extend(
        [
            {
                "tag": "action",
                "actions": [
                    {
                        "tag": "button",
                        "text": {"tag": "plain_text", "content": "Open DM/CDC Grafana"},
                        "type": "primary",
                        "url": GRAFANA_DM_URL,
                    }
                ],
            }
        ]
    )
    return {
        "config": {"wide_screen_mode": True, "enable_forward": True},
        "header": {
            "template": "red",
            "title": {
                "tag": "plain_text",
                "content": "Workflow Alert | CDC/DM Stuck Jobs",
            },
        },
        "elements": elements,
    }


def send_feishu_message(target: str, message: str) -> bool:
    token = get_feishu_token()
    if not token:
        return False

    receive_id_type, receive_id = parse_feishu_target(target)
    response = requests.post(
        f"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type={receive_id_type}",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json={
            "receive_id": receive_id,
            "msg_type": "text",
            "content": json.dumps({"text": message}, ensure_ascii=False),
        },
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(f"failed to send Feishu message: {data.get('msg')}")
    return True


def send_feishu_card(target: str, card: dict[str, Any]) -> bool:
    token = get_feishu_token()
    if not token:
        return False

    receive_id_type, receive_id = parse_feishu_target(target)
    response = requests.post(
        f"https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type={receive_id_type}",
        headers={"Authorization": f"Bearer {token}", "Content-Type": "application/json"},
        json={
            "receive_id": receive_id,
            "msg_type": "interactive",
            "content": json.dumps(card, ensure_ascii=False),
        },
        timeout=30,
    )
    response.raise_for_status()
    data = response.json()
    if data.get("code") != 0:
        raise RuntimeError(f"failed to send Feishu card: {data.get('msg')}")
    return True


def collect_anomalies() -> list[dict[str, Any]]:
    anomalies = []
    anomalies.extend(query_dm_abnormal_jobs())
    anomalies.extend(query_cdc_abnormal_jobs())
    return sorted(
        anomalies,
        key=lambda item: (
            normalize_text(item.get("component")),
            normalize_text(item.get("job_id")),
            normalize_text(item.get("status")),
        ),
    )


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Check long-stuck CDC/DM jobs")
    parser.add_argument("--state-file", default=DEFAULT_STATE_FILE)
    parser.add_argument("--format", choices=("text", "json"), default="text")
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    observed_at = dt.datetime.now(dt.UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z")
    notify_target = DEFAULT_NOTIFY_TARGET

    anomalies = collect_anomalies()
    previous_state = load_state(args.state_file)
    new_anomalies, resolved_anomalies, next_state = reconcile_state(anomalies, previous_state, observed_at)
    save_state(args.state_file, next_state)

    summary = build_summary(
        anomalies=anomalies,
        new_anomalies=new_anomalies,
        resolved_anomalies=resolved_anomalies,
        observed_at=observed_at,
    )

    if args.format == "json":
        print(json.dumps(summary, ensure_ascii=False, indent=2, sort_keys=True))
    else:
        print(format_text_summary(summary))
        if summary["should_notify"]:
            if is_true(os.environ.get("SKIP_SEND_NOTIFICATION")):
                print(f"\nFeishu notification skipped for {notify_target}")
            else:
                send_feishu_card(notify_target, build_notification_card(summary))
                print(f"\nFeishu alert card sent to {notify_target}")
        else:
            print("\nFeishu notification not sent: no new abnormal jobs")

    return 0


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