#!/usr/bin/env python3
"""Probe the public OA readiness endpoint and drive a debounced alert."""

from __future__ import annotations

import argparse
import http.client
import json
import os
import re
import socket
import ssl
import stat
import subprocess
import sys
import tempfile
import time
import urllib.error
import urllib.parse
import urllib.request
from pathlib import Path


SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
SAFE_HOST = re.compile(r"^[A-Za-z0-9.-]{1,253}$")
MAX_RESPONSE_BYTES = 64 * 1024
MAX_COUNTER = 1_000_000
MAX_TIMESTAMP = (1 << 53) - 1
STATE_SCHEMA_VERSION = 2
RECOVERY_THRESHOLD = 2
RECOVERY_WATCH_SECONDS = 60 * 60
FLAPPING_STABLE_SECONDS = 30 * 60
HEALTHY_STATUS = "PASS_APP_HEALTH_READY"
FAILURE_STATUS = "FAIL_APP_HEALTH"
FLAPPING_STATUS = "FAIL_APP_HEALTH_FLAPPING"
FAILURE_CATEGORIES = {
    "dns_resolution_failed",
    "tcp_connection_failed",
    "tls_validation_failed",
    "request_timeout",
    "http_unavailable",
    "response_contract_invalid",
    "response_too_large",
}
LEGACY_FAILURE_CATEGORIES = FAILURE_CATEGORIES | {"dns_or_connect_failed"}
PHASES = {
    "healthy",
    "failure_pending",
    "alerted",
    "recovery_pending",
    "recovery_watch",
    "flapping",
}
STATE_KEYS = {
    "schemaVersion",
    "phase",
    "activeCategory",
    "lastRecoveredCategory",
    "consecutiveFailures",
    "consecutiveSuccesses",
    "recoveredAt",
    "healthySince",
    "flapTransitions",
}


class ProbeFailure(Exception):
    def __init__(self, category: str):
        super().__init__(category)
        self.category = category


def fail(category: str) -> "NoReturn":
    print(f"[app-health][ERROR] {category}", file=sys.stderr)
    raise SystemExit(1)


def validate_regular_file(path: Path, label: str, private: bool = False) -> None:
    try:
        info = path.lstat()
    except FileNotFoundError:
        fail(f"{label}_missing")
    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
        fail(f"{label}_invalid")
    if info.st_uid != os.geteuid():
        fail(f"{label}_owner_invalid")
    if private and stat.S_IMODE(info.st_mode) != 0o600:
        fail(f"{label}_mode_invalid")
    if not private and stat.S_IMODE(info.st_mode) & 0o022:
        fail(f"{label}_mode_invalid")


def prepare_state_dir(path: Path) -> None:
    try:
        path.mkdir(parents=True, exist_ok=True, mode=0o700)
        info = path.lstat()
    except OSError:
        fail("local_state_invalid")
    if (
        stat.S_ISLNK(info.st_mode)
        or not stat.S_ISDIR(info.st_mode)
        or info.st_uid != os.geteuid()
        or stat.S_IMODE(info.st_mode) != 0o700
    ):
        fail("local_state_invalid")


def validate_ready_url(value: str) -> str:
    try:
        parsed = urllib.parse.urlsplit(value)
        port = parsed.port
    except ValueError:
        fail("ready_url_invalid")
    if (
        parsed.scheme != "https"
        or not parsed.hostname
        or not SAFE_HOST.fullmatch(parsed.hostname)
        or parsed.username is not None
        or parsed.password is not None
        or parsed.path != "/ready"
        or parsed.query
        or parsed.fragment
        or (port is not None and not 1 <= port <= 65535)
    ):
        fail("ready_url_invalid")
    return value


def new_probe_state() -> dict[str, object]:
    return {
        "schemaVersion": STATE_SCHEMA_VERSION,
        "phase": "healthy",
        "activeCategory": None,
        "lastRecoveredCategory": None,
        "consecutiveFailures": 0,
        "consecutiveSuccesses": 0,
        "recoveredAt": None,
        "healthySince": None,
        "flapTransitions": 0,
    }


def is_bounded_int(value: object, minimum: int = 0, maximum: int = MAX_COUNTER) -> bool:
    return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum


def is_category(value: object, allow_none: bool = True) -> bool:
    if value is None:
        return allow_none
    return isinstance(value, str) and value in LEGACY_FAILURE_CATEGORIES and bool(SAFE_TOKEN.fullmatch(value))


def is_timestamp(value: object, allow_none: bool = True) -> bool:
    if value is None:
        return allow_none
    return is_bounded_int(value, maximum=MAX_TIMESTAMP)


def validate_v2_state(payload: dict[str, object]) -> bool:
    if set(payload) != STATE_KEYS or payload.get("schemaVersion") != STATE_SCHEMA_VERSION:
        return False
    phase = payload.get("phase")
    active = payload.get("activeCategory")
    recovered = payload.get("lastRecoveredCategory")
    failures = payload.get("consecutiveFailures")
    successes = payload.get("consecutiveSuccesses")
    recovered_at = payload.get("recoveredAt")
    healthy_since = payload.get("healthySince")
    transitions = payload.get("flapTransitions")
    if (
        phase not in PHASES
        or not is_category(active)
        or not is_category(recovered)
        or not is_bounded_int(failures)
        or not is_bounded_int(successes)
        or not is_timestamp(recovered_at)
        or not is_timestamp(healthy_since)
        or not is_bounded_int(transitions)
    ):
        return False
    if phase == "healthy":
        return (
            active is None
            and recovered is None
            and failures == 0
            and successes == 0
            and recovered_at is None
            and healthy_since is None
            and transitions == 0
        )
    if phase == "failure_pending":
        return (
            active is not None
            and failures >= 1
            and successes == 0
            and healthy_since is None
            and transitions == 0
            and ((recovered is None and recovered_at is None) or (recovered is not None and recovered_at is not None))
        )
    if phase == "alerted":
        return (
            active is not None
            and failures >= 1
            and successes == 0
            and recovered is None
            and recovered_at is None
            and healthy_since is None
            and transitions == 0
        )
    if phase == "recovery_pending":
        return (
            active is not None
            and failures == 0
            and successes >= 1
            and recovered is None
            and recovered_at is None
            and healthy_since is None
            and transitions == 0
        )
    if phase == "recovery_watch":
        return (
            active is None
            and recovered is not None
            and failures == 0
            and successes == 0
            and recovered_at is not None
            and healthy_since is None
            and transitions == 0
        )
    if phase != "flapping" or active is None or active != recovered or recovered_at is None or transitions < 1:
        return False
    if healthy_since is None:
        return failures >= 1 and successes == 0
    return failures == 0 and successes >= 1


def migrate_v1_state(payload: dict[str, object]) -> dict[str, object] | None:
    if set(payload) != {"schemaVersion", "consecutiveFailures", "lastCategory"}:
        return None
    count = payload.get("consecutiveFailures")
    category = payload.get("lastCategory")
    if payload.get("schemaVersion") != 1 or not is_bounded_int(count) or not is_category(category):
        return None
    state = new_probe_state()
    if count == 0:
        return state if category is None else None
    if category is None:
        return None
    state.update(
        {
            "phase": "failure_pending",
            "activeCategory": category,
            "consecutiveFailures": count,
        }
    )
    return state


def load_probe_state(path: Path) -> dict[str, object]:
    if not path.exists():
        return new_probe_state()
    validate_regular_file(path, "local_state", private=True)
    try:
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        fail("local_state_invalid")
    if not isinstance(payload, dict):
        fail("local_state_invalid")
    if payload.get("schemaVersion") == STATE_SCHEMA_VERSION:
        if not validate_v2_state(payload):
            fail("local_state_invalid")
        return payload
    migrated = migrate_v1_state(payload)
    if migrated is None:
        fail("local_state_invalid")
    return migrated


def atomic_write(path: Path, payload: dict[str, object]) -> None:
    fd, temporary = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
    try:
        os.fchmod(fd, 0o600)
        with os.fdopen(fd, "w", encoding="utf-8") as handle:
            json.dump(payload, handle, ensure_ascii=True, indent=2)
            handle.write("\n")
        os.replace(temporary, path)
    except BaseException:
        try:
            os.unlink(temporary)
        except FileNotFoundError:
            pass
        raise


def capped(value: int, maximum: int = MAX_COUNTER) -> int:
    return min(value, maximum)


def elapsed(now: int, started_at: object) -> int:
    if not isinstance(started_at, int) or isinstance(started_at, bool):
        return 0
    return max(0, now - started_at)


def transition_probe_state(
    state: dict[str, object],
    category: str | None,
    now: int,
    failure_threshold: int,
) -> tuple[dict[str, object], tuple[str, str] | None]:
    """Return the next redacted state and an optional notification candidate."""
    if not validate_v2_state(state) or not is_timestamp(now, allow_none=False):
        raise ValueError("invalid transition input")
    if category is not None and category not in FAILURE_CATEGORIES:
        raise ValueError("invalid transition category")
    phase = str(state["phase"])

    if category is None:
        if phase == "healthy":
            return state.copy(), None
        if phase == "failure_pending":
            if state["recoveredAt"] is not None:
                next_state = state.copy()
                next_state.update(
                    {
                        "phase": "recovery_watch",
                        "activeCategory": None,
                        "consecutiveFailures": 0,
                    }
                )
                return next_state, None
            return new_probe_state(), None
        if phase == "alerted":
            next_state = state.copy()
            next_state.update(
                {
                    "phase": "recovery_pending",
                    "consecutiveFailures": 0,
                    "consecutiveSuccesses": 1,
                }
            )
            return next_state, None
        if phase == "recovery_pending":
            successes = capped(int(state["consecutiveSuccesses"]) + 1)
            if successes < RECOVERY_THRESHOLD:
                next_state = state.copy()
                next_state["consecutiveSuccesses"] = successes
                return next_state, None
            recovered_category = str(state["activeCategory"])
            next_state = new_probe_state()
            next_state.update(
                {
                    "phase": "recovery_watch",
                    "lastRecoveredCategory": recovered_category,
                    "recoveredAt": now,
                }
            )
            return next_state, (HEALTHY_STATUS, "public_ready_healthy")
        if phase == "recovery_watch":
            if elapsed(now, state["recoveredAt"]) >= RECOVERY_WATCH_SECONDS:
                return new_probe_state(), None
            return state.copy(), (HEALTHY_STATUS, "public_ready_healthy")
        if phase == "flapping":
            next_state = state.copy()
            if state["healthySince"] is None:
                next_state["healthySince"] = now
                next_state["consecutiveFailures"] = 0
                next_state["consecutiveSuccesses"] = 1
                next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
            else:
                next_state["consecutiveSuccesses"] = capped(int(state["consecutiveSuccesses"]) + 1)
            if elapsed(now, next_state["healthySince"]) >= FLAPPING_STABLE_SECONDS:
                return new_probe_state(), None
            return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))

    if phase == "flapping":
        next_state = state.copy()
        if state["healthySince"] is not None:
            next_state["flapTransitions"] = capped(int(state["flapTransitions"]) + 1)
        next_state.update(
            {
                "consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
                "consecutiveSuccesses": 0,
                "healthySince": None,
            }
        )
        return next_state, (FLAPPING_STATUS, str(state["activeCategory"]))

    if phase == "alerted":
        next_state = state.copy()
        next_state.update(
            {
                "activeCategory": category,
                "consecutiveFailures": capped(int(state["consecutiveFailures"]) + 1),
                "consecutiveSuccesses": 0,
            }
        )
        return next_state, (FAILURE_STATUS, category)

    if phase == "recovery_pending":
        next_state = new_probe_state()
        next_state.update(
            {
                "phase": "alerted",
                "activeCategory": category,
                "consecutiveFailures": 1,
            }
        )
        return next_state, (FAILURE_STATUS, category)

    if phase == "recovery_watch":
        within_watch = elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
        next_state = new_probe_state()
        next_state.update(
            {
                "phase": "failure_pending",
                "activeCategory": category,
                "consecutiveFailures": 1,
                "lastRecoveredCategory": state["lastRecoveredCategory"] if within_watch else None,
                "recoveredAt": state["recoveredAt"] if within_watch else None,
            }
        )
        return next_state, None

    if phase == "healthy":
        next_state = new_probe_state()
        next_state.update(
            {
                "phase": "failure_pending",
                "activeCategory": category,
                "consecutiveFailures": 1,
            }
        )
        return next_state, None

    failures = capped(int(state["consecutiveFailures"]) + 1)
    if failures < failure_threshold:
        next_state = state.copy()
        next_state.update({"activeCategory": category, "consecutiveFailures": failures})
        return next_state, None
    same_recent_failure = (
        state["lastRecoveredCategory"] == category
        and state["recoveredAt"] is not None
        and elapsed(now, state["recoveredAt"]) < RECOVERY_WATCH_SECONDS
    )
    if same_recent_failure:
        next_state = state.copy()
        next_state.update(
            {
                "phase": "flapping",
                "activeCategory": category,
                "consecutiveFailures": failures,
                "consecutiveSuccesses": 0,
                "healthySince": None,
                "flapTransitions": 1,
            }
        )
        return next_state, (FLAPPING_STATUS, category)
    next_state = new_probe_state()
    next_state.update(
        {
            "phase": "alerted",
            "activeCategory": category,
            "consecutiveFailures": failures,
        }
    )
    return next_state, (FAILURE_STATUS, category)


def classify_connection_error(reason: object) -> str:
    if isinstance(reason, (ssl.SSLError, ssl.CertificateError)):
        return "tls_validation_failed"
    if isinstance(reason, (socket.timeout, TimeoutError)):
        return "request_timeout"
    if isinstance(reason, socket.gaierror):
        return "dns_resolution_failed"
    return "tcp_connection_failed"


def classify_url_error(error: urllib.error.URLError) -> str:
    return classify_connection_error(error.reason)


def probe(ready_url: str, timeout_seconds: float, ca_file: Path | None) -> None:
    if ca_file is not None:
        validate_regular_file(ca_file, "ca_file")
    try:
        context = ssl.create_default_context(cafile=str(ca_file) if ca_file else None)
        request = urllib.request.Request(
            ready_url,
            headers={"Accept": "application/json", "User-Agent": "hyb-oa-app-health/1"},
            method="GET",
        )
        with urllib.request.urlopen(request, timeout=timeout_seconds, context=context) as response:
            if not 200 <= response.status < 300:
                raise ProbeFailure("http_unavailable")
            content_length = response.headers.get("Content-Length")
            if content_length:
                try:
                    if int(content_length) > MAX_RESPONSE_BYTES:
                        raise ProbeFailure("response_too_large")
                except ValueError:
                    raise ProbeFailure("response_contract_invalid") from None
            body = response.read(MAX_RESPONSE_BYTES + 1)
    except urllib.error.HTTPError:
        raise ProbeFailure("http_unavailable") from None
    except urllib.error.URLError as error:
        raise ProbeFailure(classify_url_error(error)) from None
    except (socket.timeout, TimeoutError):
        raise ProbeFailure("request_timeout") from None
    except (ssl.SSLError, ssl.CertificateError):
        raise ProbeFailure("tls_validation_failed") from None
    except socket.gaierror:
        raise ProbeFailure("dns_resolution_failed") from None
    except (http.client.HTTPException, ConnectionError, OSError):
        raise ProbeFailure("tcp_connection_failed") from None
    if len(body) > MAX_RESPONSE_BYTES:
        raise ProbeFailure("response_too_large")
    try:
        payload = json.loads(body.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        raise ProbeFailure("response_contract_invalid") from None
    if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict) or payload["data"].get("ok") is not True:
        raise ProbeFailure("response_contract_invalid")


def write_summary(path: Path, status: str, category: str) -> None:
    finding = status in {FAILURE_STATUS, FLAPPING_STATUS}
    payload = {
        "issueId": "APP_HEALTH",
        "status": status,
        "conclusion": status,
        "findingCount": 1 if finding else 0,
        "findings": (
            [{"source": "public_ready", "status": status, "severity": "P0", "detail": category}]
            if finding
            else []
        ),
        "privacy": "status/count/category only; no credentials, addresses, paths, or payloads",
    }
    atomic_write(path, payload)


def evaluate_alert(alert_helper: Path, webhook_file: Path, summary_file: Path, alert_state_file: Path) -> None:
    validate_regular_file(alert_helper, "alert_helper")
    try:
        result = subprocess.run(
            [
                sys.executable,
                str(alert_helper),
                "--webhook-file",
                str(webhook_file),
                "--summary-file",
                str(summary_file),
                "--state-file",
                str(alert_state_file),
            ],
            check=False,
        )
    except OSError:
        fail("alert_evaluation_failed")
    if result.returncode != 0:
        fail("alert_evaluation_failed")


def persist_transition(
    state_file: Path,
    summary_file: Path,
    alert_state_file: Path,
    state: dict[str, object],
    candidate: tuple[str, str] | None,
    alert_helper: Path,
    webhook_file: Path,
) -> None:
    try:
        atomic_write(state_file, state)
        if candidate is not None:
            status, category = candidate
            write_summary(summary_file, status, category)
    except OSError:
        fail("local_state_invalid")
    if candidate is not None:
        evaluate_alert(alert_helper, webhook_file, summary_file, alert_state_file)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--ready-url", required=True)
    parser.add_argument("--state-dir", required=True, type=Path)
    parser.add_argument("--webhook-file", required=True, type=Path)
    parser.add_argument("--alert-helper", required=True, type=Path)
    parser.add_argument("--failure-threshold", type=int, default=2)
    parser.add_argument("--timeout-seconds", type=float, default=10.0)
    parser.add_argument("--ca-file", type=Path)
    args = parser.parse_args()

    ready_url = validate_ready_url(args.ready_url)
    if not 1 <= args.failure_threshold <= 10:
        fail("failure_threshold_invalid")
    if not 0.1 <= args.timeout_seconds <= 30:
        fail("timeout_invalid")
    prepare_state_dir(args.state_dir)
    probe_state_file = args.state_dir / "app-health-probe-state.json"
    summary_file = args.state_dir / "last-app-health.json"
    alert_state_file = args.state_dir / "app-health-alert-state.json"
    state = load_probe_state(probe_state_file)
    now = int(time.time())

    try:
        probe(ready_url, args.timeout_seconds, args.ca_file)
    except ProbeFailure as error:
        try:
            next_state, candidate = transition_probe_state(state, error.category, now, args.failure_threshold)
        except ValueError:
            fail("local_state_invalid")
        persist_transition(
            probe_state_file,
            summary_file,
            alert_state_file,
            next_state,
            candidate,
            args.alert_helper,
            args.webhook_file,
        )
        alert_result = "evaluated" if candidate is not None else "pending"
        print(
            f"[app-health][ERROR] health_check_failed category={error.category} "
            f"consecutive={next_state['consecutiveFailures']} alert={alert_result}",
            file=sys.stderr,
        )
        raise SystemExit(1)

    try:
        next_state, candidate = transition_probe_state(state, None, now, args.failure_threshold)
    except ValueError:
        fail("local_state_invalid")
    persist_transition(
        probe_state_file,
        summary_file,
        alert_state_file,
        next_state,
        candidate,
        args.alert_helper,
        args.webhook_file,
    )
    print(f"[app-health][PASS] public readiness is healthy phase={next_state['phase']}")


if __name__ == "__main__":
    main()