# -*- coding: utf-8 -*-
"""Periodic health checks: Squid proxy + VPN node + egress."""
from __future__ import annotations

import json
import os
import socket
import subprocess
import time
from datetime import datetime
from pathlib import Path
from typing import Any
from urllib.parse import quote

from nodes_db import get_active, init_db, mark_probe

HOME = Path(os.environ["USERPROFILE"])
PROXY_HOME = HOME / ".claude-proxy-control"
MERGED = PROXY_HOME / "merged"
STATUS_PATH = MERGED / "monitor-status.json"
HISTORY_PATH = MERGED / "monitor-history.jsonl"
PROXY_ENV = PROXY_HOME / "proxy.env"
PROBE_DIR = PROXY_HOME / "bin"

# One binary per channel, each pinned by the builder to exactly one outbound.
#
# The monitor used to probe with plain curl.exe, which is an ordinary app subject to
# the very app-rules it was verifying. The user had curl.exe on the vpn channel, so all
# three "different" probes measured the same path and returned the same IP - and the
# squid check, finding its expectation unmet, downgraded itself to ok=True.
#
# These probes also deliberately do NOT pass -x for the squid check: Claude never sets
# a proxy either, it is sing-box that routes it. Using -x would test a path Claude
# never takes and bypass the routing under test.
PROBE_NAMES = {
    "direct": "cpc-probe-direct.exe",
    "vpn": "cpc-probe-vpn.exe",
    "squid": "cpc-probe-squid.exe",
}

# Prevent Windows/user HTTP_PROXY from poisoning "direct" curl checks
_PROXY_ENV_KEYS = (
    "HTTP_PROXY",
    "HTTPS_PROXY",
    "ALL_PROXY",
    "http_proxy",
    "https_proxy",
    "all_proxy",
)


def _now() -> str:
    return datetime.now().isoformat(timespec="seconds")


def load_proxy_env() -> dict[str, str]:
    out: dict[str, str] = {}
    if not PROXY_ENV.is_file():
        return out
    for raw in PROXY_ENV.read_text(encoding="utf-8").splitlines():
        line = raw.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        k, v = line.split("=", 1)
        out[k.strip()] = v.strip()
    return out


def tcp_ok(host: str, port: int, timeout: float = 5.0) -> bool:
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except Exception:
        return False


def _clean_env() -> dict[str, str]:
    env = dict(os.environ)
    for k in _PROXY_ENV_KEYS:
        env.pop(k, None)
    env["NO_PROXY"] = "*"
    env["no_proxy"] = "*"
    return env


def _system_curl() -> Path | None:
    cand = Path(os.environ.get("SystemRoot", r"C:\Windows")) / "System32" / "curl.exe"
    if cand.is_file():
        return cand
    from shutil import which

    found = which("curl.exe") or which("curl")
    return Path(found) if found else None


def ensure_probes() -> dict[str, str]:
    """Materialise the per-channel probe binaries. Returns {channel: exe path}.

    They are plain copies of the system curl under CPC-owned names so sing-box can
    match them by process_name. Re-copied when the source curl changes.
    """
    src = _system_curl()
    if not src:
        return {}
    out: dict[str, str] = {}
    try:
        PROBE_DIR.mkdir(parents=True, exist_ok=True)
    except Exception:
        return {}
    for channel, name in PROBE_NAMES.items():
        dst = PROBE_DIR / name
        try:
            need_copy = True
            if dst.is_file():
                s_src, s_dst = src.stat(), dst.stat()
                need_copy = s_src.st_size != s_dst.st_size
            if need_copy:
                from shutil import copyfile

                copyfile(src, dst)
            out[channel] = str(dst)
        except Exception:
            # Busy/locked copy that already exists is still usable
            if dst.is_file():
                out[channel] = str(dst)
    return out


def probes_effective() -> bool:
    """True only if the TUN running right now was started from a config that pins the probes.

    A probe binary is just an unknown process to an older TUN, so it lands in
    route.final=block and every channel looks broken. That would be reported as a leak
    and, worse, feed the builder's fail-closed logic on the next rebuild. So the probes
    are trusted only when the config on disk both matches the live TUN name and
    actually contains their rules.
    """
    try:
        st = json.loads((MERGED / "status.json").read_text(encoding="utf-8-sig"))
        cfg = json.loads((MERGED / "vpn-claude-config.json").read_text(encoding="utf-8-sig"))
    except Exception:
        return False
    tun = str(st.get("tun") or "").strip()
    if not tun:
        return False
    try:
        cfg_tun = str((cfg.get("inbounds") or [{}])[0].get("interface_name") or "")
    except Exception:
        return False
    if cfg_tun != tun:
        # Config on disk has been rebuilt but not applied - it does not describe the
        # tunnel that is actually carrying traffic.
        return False
    names: set[str] = set()
    for rule in cfg.get("route", {}).get("rules") or []:
        if not isinstance(rule, dict):
            continue
        for n in rule.get("process_name") or []:
            names.add(str(n).lower())
    return {n.lower() for n in PROBE_NAMES.values()} <= names


def curl_text(
    args: list[str],
    timeout: int = 20,
    bypass_env_proxy: bool = True,
    exe: str = "curl.exe",
) -> str:
    """curl without inheriting ambient HTTP_PROXY.

    Important: do NOT pass --noproxy '*' together with -x, or curl ignores -x too.
    """
    try:
        cmd = [exe, "-s", "-m", str(timeout)]
        # Only for "direct" checks: ignore env proxy. Explicit -x tests must not use this.
        if bypass_env_proxy and not any(a in ("-x", "--proxy") for a in args):
            cmd += ["--noproxy", "*"]
        cmd += args
        flags = getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000)
        p = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=timeout + 5,
            env=_clean_env(),
            creationflags=flags,
        )
        return (p.stdout or "").strip()
    except Exception:
        return ""


def singbox_up() -> bool:
    """Product tunnel up? Path may be hidden for elevated sing-box — use TUN/status fallbacks."""
    try:
        flags = getattr(subprocess, "CREATE_NO_WINDOW", 0x08000000)
        home = Path(os.environ.get("CLAUDE_PROXY_HOME") or (Path(os.environ["USERPROFILE"]) / ".claude-proxy-control"))
        pwsh = str(
            Path(os.environ.get("SystemRoot", r"C:\Windows"))
            / "System32"
            / "WindowsPowerShell"
            / "v1.0"
            / "powershell.exe"
        )
        # 1) any Up adapter named cpc*
        # Prefer status.json TUN name (exact), then wildcard cpc*
        tun_name = ""
        status_path = home / "merged" / "status.json"
        pid = 0
        if status_path.is_file():
            try:
                st = json.loads(status_path.read_text(encoding="utf-8-sig"))
                tun_name = str(st.get("tun") or "").strip()
                pid = int(st.get("singBoxPid") or 0)
            except Exception:
                tun_name, pid = "", 0

        tun_up = False
        if tun_name and tun_name.replace("_", "").isalnum():
            p = subprocess.run(
                [
                    pwsh,
                    "-NoProfile",
                    "-ExecutionPolicy",
                    "Bypass",
                    "-Command",
                    f"(Get-NetAdapter -Name '{tun_name}' -ErrorAction SilentlyContinue | "
                    "Select-Object -ExpandProperty Status)",
                ],
                capture_output=True,
                text=True,
                timeout=12,
                env=_clean_env(),
                creationflags=flags,
            )
            tun_up = ((p.stdout or "").strip().splitlines() or [""])[-1].strip().lower() == "up"
        if not tun_up:
            p = subprocess.run(
                [
                    pwsh,
                    "-NoProfile",
                    "-ExecutionPolicy",
                    "Bypass",
                    "-Command",
                    "$n=@(Get-NetAdapter -Name 'cpc*' -ErrorAction SilentlyContinue | "
                    "Where-Object Status -EQ 'Up'); $n.Count",
                ],
                capture_output=True,
                text=True,
                timeout=12,
                env=_clean_env(),
                creationflags=flags,
            )
            try:
                tun_up = int((p.stdout or "").strip().splitlines()[-1]) > 0
            except Exception:
                tun_up = False
        if not tun_up:
            return False

        p2 = subprocess.run(
            ["tasklist", "/FI", "IMAGENAME eq sing-box.exe", "/FO", "CSV", "/NH"],
            capture_output=True,
            text=True,
            timeout=10,
            env=_clean_env(),
            creationflags=flags,
        )
        has_sb = "sing-box.exe" in (p2.stdout or "").lower() and "info:" not in (p2.stdout or "").lower()
        if has_sb:
            return True
        if pid > 0:
            p3 = subprocess.run(
                ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"],
                capture_output=True,
                text=True,
                timeout=8,
                env=_clean_env(),
                creationflags=flags,
            )
            if str(pid) in (p3.stdout or "") and "info:" not in (p3.stdout or "").lower():
                return True
        return tun_up
    except Exception:
        return False


def run_checks() -> dict[str, Any]:
    """Run one monitoring cycle. Never stores proxy password."""
    init_db()
    env = load_proxy_env()
    host = env.get("PROXY_HOST", "")
    port = int(env.get("PROXY_PORT") or 0)
    user = env.get("PROXY_USER", "")
    password = env.get("PROXY_PASS", "")

    result: dict[str, Any] = {
        "time": _now(),
        "overall": "unknown",
        "tun_up": singbox_up(),
        "squid": {"ok": False, "detail": ""},
        "vpn_node": {"ok": False, "detail": ""},
        "egress": {"direct": "", "via_vpn_or_tun": "", "via_squid": ""},
        "active_node": None,
        "note": "",
    }

    tun_up = bool(result["tun_up"])
    probes = ensure_probes()
    pinned = tun_up and len(probes) == len(PROBE_NAMES) and probes_effective()
    result["probes"] = {
        "ready": len(probes) == len(PROBE_NAMES),
        "pinned": pinned,
        "missing": sorted(set(PROBE_NAMES) - set(probes)),
    }

    def probe(channel: str, url: str, timeout: int = 15) -> str:
        exe = probes.get(channel) if pinned else None
        if not exe:
            return curl_text([url], timeout=timeout)
        return curl_text([url], timeout=timeout, exe=exe)

    # Home ISP egress, pinned to the direct outbound. This is the reference point:
    # any other channel returning the same IP is leaking to the home connection.
    direct_eg = probe("direct", "https://ifconfig.me")
    result["egress"]["direct"] = direct_eg

    # The vpn/squid probes only mean something once the live TUN pins them.
    if pinned:
        vpn_eg = probe("vpn", "https://ifconfig.me", timeout=20)
        squid_eg = probe("squid", "https://ifconfig.me", timeout=20)
        hello = probe("squid", "https://api.anthropic.com/api/hello", timeout=25)
    else:
        vpn_eg, squid_eg, hello = "", "", ""
        if tun_up:
            # Legacy fallback: no channel-pinned measurement available, so report the
            # one egress we can see rather than pretending the channels were verified.
            vpn_eg = direct_eg
    result["egress"]["via_vpn_or_tun"] = vpn_eg
    result["egress"]["via_squid"] = squid_eg

    if not host or not port:
        result["squid"] = {"ok": False, "detail": "proxy.env неполный"}
    elif not tcp_ok(host, port, 5):
        result["squid"] = {"ok": False, "detail": f"TCP {host}:{port} недоступен"}
    elif not pinned:
        # No channel-pinned measurement available (tunnel off, or the live TUN predates
        # the probe rules). Only confirm Squid itself answers - this is the one place an
        # explicit -x request is correct, because there is no routing to test.
        proxy_url = f"http://{quote(user, safe='')}:{quote(password, safe='')}@{host}:{port}"
        hello_x = curl_text(
            ["-x", proxy_url, "https://api.anthropic.com/api/hello"],
            timeout=25,
            bypass_env_proxy=False,
        )
        why = "туннель выкл" if not tun_up else "маршруты проверок ещё не применены"
        if "hello" in hello_x.lower():
            result["squid"] = {
                "ok": True,
                "detail": f"Squid отвечает ({why} — канал не проверяется)",
            }
        else:
            result["squid"] = {"ok": False, "detail": f"Squid не отдал hello ({why})"}
    else:
        hello_ok = "hello" in hello.lower()
        if not hello_ok:
            result["squid"] = {
                "ok": False,
                "detail": f"FAIL: канал Claude не отдал hello (выход={squid_eg or '—'})",
            }
        elif not squid_eg:
            result["squid"] = {
                "ok": True,
                "detail": "hello есть, но выходной IP канала определить не удалось",
            }
        elif direct_eg and squid_eg == direct_eg:
            result["squid"] = {
                "ok": False,
                "detail": f"УТЕЧКА: канал Claude выходит с домашнего IP {squid_eg}",
            }
        else:
            result["squid"] = {
                "ok": True,
                "detail": f"OK: hello + выход {squid_eg} (дом {direct_eg or '?'})",
            }

    # Active VPN node
    active = get_active()
    if active:
        result["active_node"] = {
            "id": active["id"],
            "server": active["server"],
            "port": active["port"],
        }
        node_tcp = tcp_ok(str(active["server"]), int(active["port"]), 5)
        if not node_tcp:
            result["vpn_node"] = {
                "ok": False,
                "detail": f"TCP {active['server']}:{active['port']} недоступен",
            }
            mark_probe(int(active["id"]), False, None)
        elif pinned:
            eg = vpn_eg
            if not eg:
                result["vpn_node"] = {
                    "ok": False,
                    "detail": "TUN up, но канал VPN не отдал выходной IP",
                }
                mark_probe(int(active["id"]), False, None)
            elif direct_eg and eg == direct_eg:
                # The vpn probe is pinned to the VLESS outbound, so matching the home
                # IP means the tunnel is not actually carrying that channel.
                result["vpn_node"] = {
                    "ok": False,
                    "detail": f"УТЕЧКА: канал VPN выходит с домашнего IP {eg}",
                }
                mark_probe(int(active["id"]), False, eg)
            else:
                result["vpn_node"] = {
                    "ok": True,
                    "detail": f"TUN up, выход={eg} (дом {direct_eg or '?'})",
                }
                mark_probe(int(active["id"]), True, eg)
        else:
            why = "туннель выкл" if not tun_up else "маршруты проверок ещё не применены"
            result["vpn_node"] = {
                "ok": True,
                "detail": f"TCP OK {active['server']}:{active['port']} ({why})",
            }
    else:
        result["vpn_node"] = {"ok": False, "detail": "нет активного узла в БД"}

    squid_ok = bool(result["squid"]["ok"])
    vpn_ok = bool(result["vpn_node"]["ok"])
    if result["tun_up"]:
        if squid_ok and vpn_ok:
            result["overall"] = "ok"
            result["note"] = "Claude через Squid, остальное через VPN — оба живы"
        elif squid_ok and not vpn_ok:
            result["overall"] = "degraded"
            result["note"] = "Claude/Squid ок, VPN плохо — обнови узел (Sota Connect→Disconnect)"
        elif not squid_ok and vpn_ok:
            result["overall"] = "degraded"
            result["note"] = "VPN ок, Squid плохо — Claude может не отвечать"
        else:
            result["overall"] = "fail"
            result["note"] = "И Squid, и VPN плохие"
    else:
        # Idle = expected. Never overall=fail just because Squid probe failed while TUN is off —
        # full Squid path is validated after Start; don't scare the user before launch.
        if active:
            result["overall"] = "idle_ok"
            result["note"] = "Туннель выкл — нормально; жми «Запустить Claude»"
        else:
            result["overall"] = "idle_warn"
            result["note"] = "Туннель выкл; сначала поймай/выбери узел Sota"

    MERGED.mkdir(parents=True, exist_ok=True)
    STATUS_PATH.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    with HISTORY_PATH.open("a", encoding="utf-8") as f:
        f.write(
            json.dumps(
                {
                    "time": result["time"],
                    "overall": result["overall"],
                    "tun_up": result["tun_up"],
                    "squid_ok": result["squid"]["ok"],
                    "vpn_ok": result["vpn_node"]["ok"],
                    "egress": result["egress"]["via_vpn_or_tun"] or result["egress"]["direct"],
                    "squid_egress": result["egress"]["via_squid"],
                    "node": result.get("active_node"),
                    "note": result.get("note"),
                },
                ensure_ascii=False,
            )
            + "\n"
        )
    return result


def load_last() -> dict[str, Any]:
    if not STATUS_PATH.is_file():
        return {}
    try:
        return json.loads(STATUS_PATH.read_text(encoding="utf-8"))
    except Exception:
        return {}


def loop(interval_sec: int = 60) -> None:
    while True:
        try:
            r = run_checks()
            print(
                f"{r['time']} overall={r['overall']} squid={r['squid']['ok']} "
                f"vpn={r['vpn_node']['ok']} tun={r['tun_up']} eg={r['egress'].get('direct')}"
            )
        except Exception as e:
            print(f"monitor error: {e}")
        time.sleep(interval_sec)


if __name__ == "__main__":
    import sys

    if len(sys.argv) > 1 and sys.argv[1] == "--loop":
        interval = int(sys.argv[2]) if len(sys.argv) > 2 else 60
        loop(interval)
    else:
        r = run_checks()
        print(
            json.dumps(
                {
                    k: r[k]
                    for k in (
                        "time",
                        "overall",
                        "note",
                        "tun_up",
                        "squid",
                        "vpn_node",
                        "egress",
                        "active_node",
                    )
                },
                ensure_ascii=False,
                indent=2,
            )
        )
