"""Проверка внешней доступности сервисов (задача #521).

Проба идёт по публичному пути, а не через локальный DNS:

    публичный резолвер → публичный IP → TLS с SNI и проверкой сертификата →
    GET публичного пути → код ответа

Так ловятся поломки, которых не видит внутренний health-чек
(`services_probe.py`): публичный DNS ведёт не на тот адрес, сертификат
истёк или выдан другому имени, порт закрыт снаружи, reverse proxy отвечает
5xx. Проба запускается с самой панели, поэтому внешний файрвол ПЕРЕД панелью
она не проверяет — это ограничение, а не баг.

Состояния точки:

    up            DNS ок (или ожидаемый IP не задан), TLS ок, код < 500 и не 421
    dns_mismatch  ожидаемый IP не входит в ответ публичного DNS
    dns_fail      NXDOMAIN / нет A-AAAA / пустой ответ
    tls_fail      TLS-рукопожатие или проверка сертификата не прошли
    http_fail     нет/битая строка статуса, код >= 500 или 421
    down          TCP: отказ, сброс, таймаут
    probe_error   недоступен сам резолвер — о сервисе ничего не известно,
                  поэтому такой замер НЕ участвует в алертах

Свежесть точки и «pending» выводит API `GET /services`. История — 7 дней,
как у внутренних проб.
"""

import asyncio
import ipaddress
import ssl
import time
from datetime import datetime, timedelta, timezone
from urllib.parse import urlparse

import dns.asyncresolver
import dns.exception
import dns.resolver

from app.config import get_settings
from app.db import get_db, j
from app.events import record_event

RETENTION_DAYS = 7
PORT = 443
# пауза на закрытие TLS-соединения: wait_closed() на битом shutdown может висеть
CLOSE_TIMEOUT_S = 0.5

_warned_bad_ip = False


def _expected_ip() -> str:
    """GHARD_PUBLIC_IP, если это валидный IP; иначе '' (сверка DNS выключена).

    Сломанную настройку не роняем на каждой волне: предупреждаем один раз.
    """
    global _warned_bad_ip
    value = get_settings().public_ip.strip()
    if not value:
        return ""
    try:
        ipaddress.ip_address(value)
    except ValueError:
        if not _warned_bad_ip:
            print(
                f"external probe: GHARD_PUBLIC_IP={value!r} не IP — сверка DNS отключена",
                flush=True,
            )
            _warned_bad_ip = True
        return ""
    return value


def _resolver():
    """Резолвер на публичный DNS: локальный split-horizon для сверки не годится.

    Создаётся внутри работающей петли (не на импорте): так его не привязывает
    ни один завершившийся event loop.
    """
    settings = get_settings()
    resolver = dns.asyncresolver.Resolver(configure=False)
    resolver.nameservers = [settings.public_resolver]
    resolver.timeout = settings.external_timeout
    resolver.lifetime = settings.external_timeout
    return resolver


def _short(exc: Exception, limit: int = 200) -> str:
    text = str(exc) or type(exc).__name__
    return text.split(";")[0][:limit]


async def _resolve(host: str, resolver, timeout: float) -> tuple[list[str], str, str]:
    """(ips, state, message); state '' — ответ получен и непуст.

    Семейство запросов выбирается по ожидаемому IP: незачем спрашивать AAAA,
    если площадка IPv4, и наоборот. Без ожидаемого IP спрашиваем обе семьи.
    """
    expected = _expected_ip()
    if expected and ":" in expected:
        rdtypes = ("AAAA",)
    elif expected:
        rdtypes = ("A",)
    else:
        rdtypes = ("A", "AAAA")

    answers: list[str] = []
    resolver_failed = False
    last_exc: Exception | None = None
    for rdtype in rdtypes:
        try:
            answer = await resolver.resolve(host, rdtype, search=False, lifetime=timeout)
        except dns.resolver.NXDOMAIN:
            return [], "dns_fail", "NXDOMAIN: домена нет"
        except dns.resolver.NoAnswer:
            continue  # записей этого семейства нет — проверим следующее
        except dns.exception.DNSException as exc:
            # Timeout/Lifetime/NoNameservers: резолвер не дал ответа, поэтому
            # судить о сервисе нельзя (NoNameservers приходит и при блокировке
            # UDP/53, и при отказавших NS — оба случая не про сервис)
            resolver_failed = True
            last_exc = exc
            continue
        answers.extend(str(record) for record in answer)

    if answers:
        return answers, "", ""
    if resolver_failed:
        name = type(last_exc).__name__ if last_exc is not None else "DNSException"
        return [], "probe_error", f"резолвер {get_settings().public_resolver} не ответил ({name})"
    return [], "dns_fail", "нет A/AAAA записей"


def _parse_status(line: bytes) -> tuple[str, int | None, str]:
    """Строка статуса HTTP/1.1 → (state, code, message)."""
    if not line:
        return "http_fail", None, "пустой ответ"
    try:
        text = line.decode("latin-1").strip()
        _version, code_text, *_rest = text.split(" ", 2)
        code = int(code_text)
    except (ValueError, UnicodeError):
        return "http_fail", None, f"битая строка статуса: {line[:60]!r}"
    if code == 421:
        return "http_fail", code, "421 (SNI/Host не совпал с виртуальным хостом)"
    if code >= 500:
        return "http_fail", code, f"http {code}"
    # 3xx/4xx — площадка достижима; редиректы не идём: важен именно этот IP и SNI
    return "up", code, ""


async def _tls_http_get(ip: str, host: str, target: str, timeout: float) -> dict:
    """TLS с проверкой сертификата + один GET; читаем только строку статуса."""
    started = time.monotonic()
    state = "down"
    code: int | None = None
    cert_days: int | None = None
    message = ""
    writer = None
    try:
        context = ssl.create_default_context()
        context.set_alpn_protocols(["http/1.1"])  # h2 мы разбирать не умеем
        async with asyncio.timeout(timeout):
            reader, writer = await asyncio.open_connection(
                ip,
                PORT,
                ssl=context,
                server_hostname=host,
                ssl_handshake_timeout=timeout,
            )
            ssl_object = writer.get_extra_info("ssl_object")
            cert = ssl_object.getpeercert() if ssl_object is not None else {}
            if cert and cert.get("notAfter"):
                cert_days = int(
                    (ssl.cert_time_to_seconds(cert["notAfter"]) - time.time()) // 86400
                )
            request = (
                f"GET {target} HTTP/1.1\r\n"
                f"Host: {host}\r\n"
                "User-Agent: ghard-panel-external/1\r\n"
                "Accept: */*\r\n"
                "Connection: close\r\n\r\n"
            )
            writer.write(request.encode("ascii", "replace"))
            await writer.drain()
            line = await reader.readline()
            state, code, message = _parse_status(line)
    except ssl.SSLCertVerificationError as exc:
        state, message = "tls_fail", f"сертификат: {_short(exc)}"
    except ssl.SSLError as exc:
        state, message = "tls_fail", f"tls: {_short(exc)}"
    except TimeoutError:
        state, message = "down", "таймаут соединения"
    except OSError as exc:
        state, message = "down", _short(exc)
    finally:
        if writer is not None:
            try:
                writer.close()
                async with asyncio.timeout(CLOSE_TIMEOUT_S):
                    await writer.wait_closed()
            except (TimeoutError, OSError, ssl.SSLError, asyncio.CancelledError):
                pass
    return {
        "state": state,
        "code": code,
        "latency_ms": round((time.monotonic() - started) * 1000, 1),
        "cert_days": cert_days,
        "message": message,
    }


def _tls_note(tls: dict | None) -> str:
    """Вторичный результат TLS для `message` (не терять причину отказа)."""
    if tls is None:
        return ""
    if tls["state"] == "up":
        return "tls/HTTP ок"
    return f"tls/HTTP: {tls['message'] or tls['state']}"


def _classify(
    dns_state: str,
    dns_message: str,
    mismatch: bool,
    ips: list[str],
    expected: str,
    tls: dict | None,
) -> tuple[str, str]:
    """Итоговое состояние пробы и человекочитаемая причина.

    Порядок: недоступный резолвер → DNS-ошибка → расхождение DNS → TLS →
    HTTP. Вторичный результат добавляется к сообщению, чтобы причина не
    терялась (например, «DNS не туда, но TLS к ожидаемому IP жив»).
    """
    if dns_state == "probe_error":
        return "probe_error", dns_message
    if dns_state == "dns_fail":
        note = _tls_note(tls)
        return "dns_fail", f"{dns_message} ({note})" if note else dns_message
    if mismatch:
        found = ", ".join(ips) or "—"
        note = _tls_note(tls)
        text = f"DNS: {found}, ожидался {expected}"
        return "dns_mismatch", f"{text} ({note})" if note else text
    if tls is None:
        return "probe_error", dns_message or "проба не выполнена"
    if tls["state"] != "up":
        return tls["state"], tls["message"]
    return "up", ""


async def probe_once(service: dict, resolver) -> dict:
    """Одна внешняя проба сервиса → {state, code, latency_ms, dns_ips,
    cert_days, message}.

    С ожидаемым IP TLS не зависит от DNS (идём на IP) — поэтому сверка DNS и
    запрос выполняются параллельно и укладываются в один бюджет. Без
    ожидаемого IP соединение возможно только после резолва, бюджет делится
    пополам.
    """
    settings = get_settings()
    parsed = urlparse(service["public_url"])
    host = parsed.hostname or ""
    target = parsed.path or "/"
    if parsed.query:
        target = f"{target}?{parsed.query}"
    expected = _expected_ip()

    if expected:
        dns_result, tls = await asyncio.gather(
            _resolve(host, resolver, settings.external_timeout),
            _tls_http_get(expected, host, target, settings.external_timeout),
        )
    else:
        dns_result = await _resolve(host, resolver, settings.external_timeout / 2)
        tls = None
        if not dns_result[1]:
            tls = await _tls_http_get(
                dns_result[0][0], host, target, settings.external_timeout / 2
            )

    ips, dns_state, dns_message = dns_result
    mismatch = bool(expected) and not dns_state and expected not in ips
    state, message = _classify(dns_state, dns_message, mismatch, ips, expected, tls)
    return {
        "state": state,
        "code": tls["code"] if tls is not None else None,
        "latency_ms": tls["latency_ms"] if tls is not None else 0.0,
        "dns_ips": ips,
        "cert_days": tls["cert_days"] if tls is not None else None,
        "message": message,
    }


async def _internal_state(db, service_id: int) -> str:
    """Последнее внутреннее состояние сервиса — для сообщения об алерте."""
    cursor = await db.execute(
        "SELECT state FROM service_samples WHERE service_id = ? ORDER BY id DESC LIMIT 1",
        (service_id,),
    )
    row = await cursor.fetchone()
    return row["state"] if row is not None else ""


def _event_data(service: dict, probe: dict, internal_state: str) -> dict:
    data = {
        "service_id": service["id"],
        "name": service["name"],
        "public_url": service["public_url"],
        "state": probe["state"],
        "message": probe["message"],
        "dns_ips": probe["dns_ips"],
        "internal_state": internal_state,
    }
    if probe.get("code") is not None:
        data["code"] = probe["code"]
    if probe.get("cert_days") is not None:
        data["cert_days"] = probe["cert_days"]
    return data


async def _alert_step(
    db, service: dict, probe: dict, last_ts: str | None, now: datetime
) -> None:
    """Конечный автомат алертов: одно событие на инцидент, recovered — только
    после реального down-события (короткое моргание Synapse не будит).

    Состояние живёт в колонках services (`external_bad_since` /
    `external_alert_open`) и переживает рестарт панели. Если между замерами
    был разрыв больше 3 интервалов, сервис не наблюдался — окно отказа
    начинается заново, чтобы не приписать ему падение, которого мы не видели.
    Открытый алерт при этом не сбрасываем: повторного down не будет.
    """
    state = probe["state"]
    service_id = service["id"]
    if state == "probe_error":  # о сервисе ничего не известно
        return
    if state == "up":
        if service["external_alert_open"]:
            since = _parse_ts(service["external_bad_since"])
            minutes = max(1, int((now - since).total_seconds() // 60)) if since else 0
            message = f"{service['name']} снова доступен снаружи"
            if minutes:
                message += f" (отказ длился {minutes} мин)"
            await record_event(
                db,
                type="service_external_recovered",
                severity="info",
                message=message,
                data=_event_data(service, probe, await _internal_state(db, service_id)),
            )
            await db.execute(
                "UPDATE services SET external_alert_open = 0, external_bad_since = NULL"
                " WHERE id = ?",
                (service_id,),
            )
        elif service["external_bad_since"] is not None:
            await db.execute(
                "UPDATE services SET external_bad_since = NULL WHERE id = ?", (service_id,)
            )
        return

    settings = get_settings()
    bad_since = _parse_ts(service["external_bad_since"])
    if bad_since is not None and last_ts is not None:
        last = _parse_ts(last_ts)
        if last is not None and (now - last).total_seconds() > settings.external_interval * 3:
            bad_since = None
    if bad_since is None:
        bad_since = now
        await db.execute(
            "UPDATE services SET external_bad_since = ? WHERE id = ?",
            (bad_since.isoformat(), service_id),
        )
    if service["external_alert_open"]:
        return
    if (now - bad_since).total_seconds() < settings.external_alert_after:
        return

    internal_state = await _internal_state(db, service_id)
    minutes = max(1, int((now - bad_since).total_seconds() // 60))
    message = (
        f"{service['name']} недоступен снаружи {minutes} мин"
        f" ({state}: {probe['message']})"
    )
    if internal_state == "up":
        message += " — внутри сервис отвечает"
    await record_event(
        db,
        type="service_external_down",
        severity="warning",
        message=message,
        data=_event_data(service, probe, internal_state),
    )
    await db.execute(
        "UPDATE services SET external_alert_open = 1 WHERE id = ?", (service_id,)
    )


def _parse_ts(value: str | None) -> datetime | None:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value)
    except ValueError:
        return None
    return parsed if parsed.tzinfo else parsed.replace(tzinfo=timezone.utc)


async def _last_ts(db, service_ids: list[int]) -> dict[int, str]:
    """ts последней внешней точки до вставки новой."""
    result: dict[int, str] = {}
    for service_id in service_ids:
        cursor = await db.execute(
            "SELECT ts FROM service_external_samples WHERE service_id = ?"
            " ORDER BY id DESC LIMIT 1",
            (service_id,),
        )
        row = await cursor.fetchone()
        if row is not None:
            result[service_id] = row["ts"]
    return result


def _fallback_probe(exc: Exception) -> dict:
    """Упавшая проба одного сервиса: волну не роняем, пишем probe_error."""
    return {
        "state": "probe_error",
        "code": None,
        "latency_ms": 0.0,
        "dns_ips": [],
        "cert_days": None,
        "message": _short(exc),
    }


async def sample_once() -> int:
    """Одна волна внешних проб по всем сервисам с публичным URL."""
    db = get_db()
    cursor = await db.execute(
        "SELECT id, name, public_url, external_bad_since, external_alert_open"
        " FROM services WHERE public_url != ''"
    )
    rows = await cursor.fetchall()
    if not rows:
        return 0

    services = [dict(row) for row in rows]
    resolver = _resolver()
    last_ts = await _last_ts(db, [service["id"] for service in services])
    results = await asyncio.gather(
        *(probe_once(service, resolver) for service in services),
        return_exceptions=True,
    )

    now = datetime.now(timezone.utc)
    ts = now.isoformat()
    for service, result in zip(services, results):
        probe = _fallback_probe(result) if isinstance(result, Exception) else result
        await db.execute(
            "INSERT INTO service_external_samples"
            " (service_id, ts, state, dns_ips, code, latency_ms, cert_days, message)"
            " VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
            (
                service["id"],
                ts,
                probe["state"],
                j(probe["dns_ips"]),
                probe["code"],
                probe["latency_ms"],
                probe["cert_days"],
                probe["message"],
            ),
        )
        await _alert_step(db, service, probe, last_ts.get(service["id"]), now)

    cutoff = (now - timedelta(days=RETENTION_DAYS)).isoformat()
    await db.execute("DELETE FROM service_external_samples WHERE ts < ?", (cutoff,))
    await db.commit()
    return len(services)


async def probe_service_once(service_id: int) -> dict | None:
    """Немедленная проба после создания сервиса или смены публичного URL."""
    db = get_db()
    cursor = await db.execute(
        "SELECT id, name, public_url, external_bad_since, external_alert_open"
        " FROM services WHERE id = ?",
        (service_id,),
    )
    row = await cursor.fetchone()
    if row is None or not row["public_url"]:
        return None

    service = dict(row)
    last_ts = (await _last_ts(db, [service_id])).get(service_id)
    probe = await probe_once(service, _resolver())
    now = datetime.now(timezone.utc)
    await db.execute(
        "INSERT INTO service_external_samples"
        " (service_id, ts, state, dns_ips, code, latency_ms, cert_days, message)"
        " VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
        (
            service_id,
            now.isoformat(),
            probe["state"],
            j(probe["dns_ips"]),
            probe["code"],
            probe["latency_ms"],
            probe["cert_days"],
            probe["message"],
        ),
    )
    await _alert_step(db, service, probe, last_ts, now)
    await db.commit()
    return probe


async def loop() -> None:
    interval = get_settings().external_interval
    while True:
        try:
            await sample_once()
        except Exception as exc:  # не роняем цикл из-за одной ошибки
            print(f"external probe error: {exc}", flush=True)
        await asyncio.sleep(interval)