#!/usr/bin/env python3
"""Send a Feishu alert when the prod Lead incremental sync workflow fails.

Reads the redacted evidence artifact produced by
``build_incremental_sync_evidence.py`` so the alert can name the stalled window
and the failing stage without anyone opening the run first. Falls back to a
minimal alert when the run failed before the evidence artifact existed.
"""
from __future__ import annotations

import argparse
import json
import os
import re
from pathlib import Path
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen

from salesforce_migration.incremental_sync_state import (
    sanitize_error_category_counts,
)

ANSI_RE = re.compile(r"\x1b\[[0-9;?]*[ -/]*[@-~]")
# The evidence artifact's producer defines it as PII-free. Console excerpts
# must apply the same treatment to Salesforce ids, HubSpot ids, and e-mail
# addresses. Sync failures routinely quote SQL carrying id lists, and a
# chat channel is a far weaker boundary than a 3-day run artifact. Triage needs
# the error shape and its counts, never the identifiers themselves.
EMAIL_RE = re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}")
UUID_RE = re.compile(r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b")
# Salesforce record ids are 15 or 18 alphanumeric chars and always contain a
# digit, which keeps ordinary words and identifiers out of the match.
SALESFORCE_ID_RE = re.compile(r"\b(?=[A-Za-z0-9]*\d)[A-Za-z0-9]{18}\b|\b(?=[A-Za-z0-9]*\d)[A-Za-z0-9]{15}\b")
HUBSPOT_ID_RE = re.compile(r"\bhubspot[-_]?\d+\b", re.IGNORECASE)
# Production HubSpot ids are bare numbers (e.g. 239460751165 in the sync tests),
# so length is the only signal available. Nine digits stays clear of everything
# triage reads — counts, row totals, mysql codes like ERROR 1054, exit codes —
# while covering the 11-13 digit ids HubSpot actually issues.
NUMERIC_ID_RE = re.compile(r"\b\d{9,}\b")
# Collapse the id lists SQL errors emit so one masked run does not fill the card.
REPEATED_MASK_RE = re.compile(r"(?:<(?:sfid|uuid|id|hubspot-id)>[,'\s]*){2,}")
ERROR_LINE_RE = re.compile(
    r"(^|\s)("
    r"[A-Za-z_.]*(?:Error|Exception):"
    r"|ERROR \d+ \([0-9A-Z]+\)"
    r"|\[warn\]"
    r"|##\[error\]"
    r")",
)
MAX_DETAIL_LINES = 12
MAX_DETAIL_LINE_CHARS = 500
MAX_DETAIL_CHARS = 3000


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--app-id", required=True)
    parser.add_argument("--app-secret", default="")
    parser.add_argument(
        "--app-secret-env",
        default="",
        help="Environment variable holding the app secret; keeps it out of the process list.",
    )
    parser.add_argument("--chat-id", required=True)
    parser.add_argument("--run-url", required=True)
    parser.add_argument("--evidence-json", default="")
    parser.add_argument("--console-log", default="")
    parser.add_argument("--run-attempt", default="")
    parser.add_argument("--trigger", default="")
    return parser.parse_args()


def post_json(url: str, payload: dict[str, object], headers: dict[str, str] | None = None) -> dict[str, object]:
    merged_headers = {"Content-Type": "application/json; charset=utf-8"}
    if headers:
        merged_headers.update(headers)
    request = Request(
        url,
        data=json.dumps(payload).encode("utf-8"),
        headers=merged_headers,
        method="POST",
    )
    try:
        with urlopen(request, timeout=30) as response:
            return json.loads(response.read().decode("utf-8"))
    except HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code} calling {url}: {body}") from exc
    except URLError as exc:
        raise RuntimeError(f"network error calling {url}: {exc}") from exc


def fetch_tenant_access_token(app_id: str, app_secret: str) -> str:
    payload = post_json(
        "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal",
        {"app_id": app_id, "app_secret": app_secret},
    )
    token = str(payload.get("tenant_access_token", "")).strip()
    if not token:
        raise RuntimeError(
            f"missing tenant_access_token in Feishu response: {json.dumps(payload, ensure_ascii=True)}"
        )
    return token


def redact(text: str) -> str:
    """Mask every record identifier before an excerpt leaves the run.

    Keeps the diagnostic shape ("conflict_count=1", the exception type) while
    dropping the values that identify people or records.
    """
    masked = EMAIL_RE.sub("***@***", text)
    masked = UUID_RE.sub("<uuid>", masked)
    masked = HUBSPOT_ID_RE.sub("<hubspot-id>", masked)
    masked = SALESFORCE_ID_RE.sub("<sfid>", masked)
    masked = NUMERIC_ID_RE.sub("<id>", masked)
    return REPEATED_MASK_RE.sub("<ids...> ", masked)


def load_evidence(path: str) -> dict[str, object]:
    if not path:
        return {}
    evidence_path = Path(path).expanduser()
    if not evidence_path.is_file():
        return {}
    try:
        payload = json.loads(evidence_path.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}
    return payload if isinstance(payload, dict) else {}


def failed_window(evidence: dict[str, object]) -> dict[str, object]:
    windows = evidence.get("windows")
    if not isinstance(windows, list):
        return {}
    for window in windows:
        if isinstance(window, dict) and str(window.get("status", "")) == "failed":
            return window
    return {}


def summarize_evidence(evidence: dict[str, object]) -> dict[str, str]:
    """Flatten the evidence artifact into the fields the card renders."""
    if not evidence:
        # No artifact at all — the card carries its own explanation for this.
        return {}
    run = evidence.get("run") if isinstance(evidence.get("run"), dict) else {}
    catchup = evidence.get("catchup") if isinstance(evidence.get("catchup"), dict) else {}
    durable = (
        evidence.get("durable_state") if isinstance(evidence.get("durable_state"), dict) else {}
    )
    window = failed_window(evidence)

    summary: dict[str, str] = {}
    mode = str(run.get("mode", "")).strip()
    if mode:
        summary["Mode"] = mode

    # No catch-up summary means the run died before any window was attempted
    # (e.g. preflight). Reporting a cursor state there would read as "a window
    # failed and stalled the lane", which is the wrong incident.
    if evidence.get("catchup_summary_found") is not True:
        summary["Stage"] = "failed before any window was attempted"
        return summary

    if window:
        summary["Failed window"] = (
            f"{window.get('window_start', '?')} .. {window.get('window_end', '?')}"
        )
        stage = str(window.get("failure_stage", "")).strip()
        error_type = str(window.get("error_type", "")).strip()
        if stage and stage != "unknown":
            summary["Failure stage"] = stage
        if error_type and error_type != "unknown":
            summary["Error type"] = error_type
        replay_status = str(window.get("replay_status", "")).strip()
        if replay_status and replay_status != "unknown":
            def count(name: str) -> int:
                try:
                    return max(0, int(window.get(name, 0) or 0))
                except (TypeError, ValueError):
                    return 0

            summary["Replay"] = (
                f"{replay_status}; attempts={count('replay_attempt_count')}; "
                "consecutive_no_progress="
                f"{count('replay_consecutive_no_progress_count')}; "
                f"remaining={count('replay_remaining_candidate_count')}"
            )
        raw_categories = window.get("replay_error_category_counts")
        categories = sanitize_error_category_counts(raw_categories)
        if categories:
            summary["Replay errors"] = ", ".join(
                f"{category}={categories[category]}" for category in categories
            )
    elif catchup.get("effective_start"):
        summary["Planned range"] = (
            f"{catchup.get('effective_start', '?')} .. {catchup.get('effective_until', '?')}"
        )

    if durable:
        # A frozen cursor is the operationally important part: scheduled runs
        # will keep retrying the same window until someone intervenes.
        summary["Cursor"] = (
            "frozen (no window committed)"
            if durable.get("state_unchanged") is True
            else "advanced before failure"
        )
    return summary


def format_console_errors(raw: str) -> str:
    text = ANSI_RE.sub("", raw or "")
    text = text.replace("\r\n", "\n").replace("\r", "\n")
    lines: list[str] = []
    for raw_line in text.splitlines():
        line = re.sub(r"[ \t]+", " ", raw_line).strip()
        if not line or not ERROR_LINE_RE.search(line):
            continue
        if len(line) > MAX_DETAIL_LINE_CHARS:
            line = line[: MAX_DETAIL_LINE_CHARS - 3].rstrip() + "..."
        lines.append(line)
    # The decisive error is emitted last, so keep the tail rather than the head.
    rendered = "\n".join(lines[-MAX_DETAIL_LINES:])
    if len(rendered) > MAX_DETAIL_CHARS:
        rendered = rendered[-MAX_DETAIL_CHARS:]
    return redact(rendered)


def read_console_log(path: str) -> str:
    if not path:
        return ""
    log_path = Path(path).expanduser()
    if not log_path.is_file():
        return ""
    try:
        return log_path.read_text(encoding="utf-8", errors="replace")
    except OSError:
        return ""


def build_card(
    *,
    summary: dict[str, str],
    console_errors: str,
    run_url: str,
    run_attempt: str,
    trigger: str,
    evidence_found: bool,
) -> dict[str, object]:
    fields = [
        {"is_short": True, "text": {"tag": "lark_md", "content": f"**{label}**\n{value}"}}
        for label, value in summary.items()
    ]
    if trigger:
        fields.append(
            {"is_short": True, "text": {"tag": "lark_md", "content": f"**Trigger**\n{trigger}"}}
        )
    if run_attempt:
        fields.append(
            {"is_short": True, "text": {"tag": "lark_md", "content": f"**Attempt**\n{run_attempt}"}}
        )

    elements: list[dict[str, object]] = []
    if fields:
        elements.append({"tag": "div", "fields": fields})
    if not evidence_found:
        elements.append(
            {
                "tag": "div",
                "text": {
                    "tag": "lark_md",
                    "content": "No evidence artifact was produced; the run failed before the sync stage.",
                },
            }
        )
    if console_errors:
        elements.extend(
            [
                {"tag": "hr"},
                {"tag": "div", "text": {"tag": "plain_text", "content": console_errors}},
            ]
        )
    elements.extend(
        [
            {"tag": "hr"},
            {"tag": "div", "text": {"tag": "lark_md", "content": f"[Open workflow run]({run_url})"}},
        ]
    )
    return {
        "config": {"wide_screen_mode": True},
        "header": {
            "template": "red",
            "title": {
                "tag": "plain_text",
                "content": "Salesforce Prod Lead incremental sync failed",
            },
        },
        "elements": elements,
    }


def send_chat_message(token: str, chat_id: str, card: dict[str, object]) -> None:
    payload = {
        "receive_id": chat_id,
        "msg_type": "interactive",
        "content": json.dumps(card, ensure_ascii=False),
    }
    response = post_json(
        "https://open.feishu.cn/open-apis/im/v1/messages?receive_id_type=chat_id",
        payload,
        headers={"Authorization": f"Bearer {token}"},
    )
    if response.get("code") not in (0, "0", None):
        raise RuntimeError(f"Feishu send message failed: {json.dumps(response, ensure_ascii=True)}")


def resolve_app_secret(args: argparse.Namespace) -> str:
    secret_env = str(args.app_secret_env or "").strip()
    secret = os.environ.get(secret_env, "") if secret_env else str(args.app_secret or "")
    if not secret:
        raise ValueError("Feishu app secret is required")
    return secret


def main() -> int:
    args = parse_args()
    app_secret = resolve_app_secret(args)
    evidence = load_evidence(args.evidence_json)
    card = build_card(
        summary=summarize_evidence(evidence),
        console_errors=format_console_errors(read_console_log(args.console_log)),
        run_url=args.run_url,
        run_attempt=args.run_attempt,
        trigger=args.trigger,
        evidence_found=bool(evidence),
    )
    token = fetch_tenant_access_token(args.app_id, app_secret)
    send_chat_message(token, args.chat_id, card)
    return 0


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