"""Read-only runtime collection and append-only observation persistence."""

from __future__ import annotations

import fcntl
import hashlib
import ipaddress
import json
import os
import subprocess
from datetime import date, datetime
from pathlib import Path
from typing import Final

import yaml
from pydantic import JsonValue, TypeAdapter

from .nutricoach_v150_observer_contract import (
    CANDIDATE,
    Observation,
    RuntimeSnapshot,
    WINDOW_END,
    WINDOW_START,
)

PROFILE: Final = Path("/home/cube/.hermes/profiles/dualcoachtest")
RUNTIME: Final = PROFILE / ".strict-runtime/81a7a06e-v150"
MIGRATION: Final = Path("/home/cube/.hermes/migrations/nutricoach-v1.5.0-combined")
EXECUTION: Final = (
    MIGRATION / "live-executions-v15/nutricoach-v150-v15-runtime-authority-r70"
)
AUTHORIZATION: Final = (
    MIGRATION / "live-authorization-v15/nutricoach-v150-v15-runtime-authority-r70"
)
OUTPUT: Final = MIGRATION / "observer-r70"
SERVICE: Final = "hermes-gateway-dualcoachtest.service"
_OBJECT: Final = TypeAdapter(dict[str, JsonValue])
_BAD_JOURNAL: Final = (
    "failed to connect",
    "Task26 service activation authority rejected",
    "Telegram startup failed",
    "Traceback (most recent call last)",
)
_TELEGRAM_NETWORKS: Final = (
    ipaddress.ip_network("149.154.160.0/20"),
    ipaddress.ip_network("91.108.4.0/22"),
    ipaddress.ip_network("91.108.8.0/22"),
    ipaddress.ip_network("91.108.12.0/22"),
    ipaddress.ip_network("91.108.16.0/22"),
)


def _run(command: tuple[str, ...]) -> str:
    return subprocess.run(
        command,
        check=True,
        capture_output=True,
        text=True,
    ).stdout


def _mapping(value: JsonValue | None, label: str) -> dict[str, JsonValue]:
    if not isinstance(value, dict):
        raise TypeError(label)
    return value


def _systemd() -> dict[str, str]:
    output = _run((
        "systemctl",
        "--user",
        "show",
        SERVICE,
        "--property=ActiveState",
        "--property=SubState",
        "--property=MainPID",
        "--property=ExecStart",
        "--property=NRestarts",
    ))
    values: dict[str, str] = {}
    for line in output.splitlines():
        key, separator, value = line.partition("=")
        if not separator:
            raise ValueError("systemd output")
        values[key] = value
    return values


def _config_health() -> tuple[bool, bool, int, bool]:
    document = _OBJECT.validate_python(
        yaml.safe_load((PROFILE / "config.yaml").read_text(encoding="utf-8"))
    )
    platforms = _mapping(document.get("platforms"), "platforms")
    telegram = _mapping(platforms.get("telegram"), "telegram")
    extra = _mapping(telegram.get("extra"), "extra")
    nutrition = _mapping(extra.get("nutrition_coaching"), "nutrition")
    weekly = _mapping(nutrition.get("weekly_operations"), "weekly")
    preflight = _mapping(extra.get("production_preflight"), "preflight")
    registry = _OBJECT.validate_json((PROFILE / "customers/registry.json").read_bytes())
    policy = _mapping(registry.get("admission_policy"), "admission_policy")
    candidate_matches = (
        preflight.get("candidate_package_identity") == CANDIDATE
        and policy.get("candidate_digest") == CANDIDATE
    )
    inbox = extra.get("channel_inbox")
    inbox_off = inbox is None or (
        isinstance(inbox, dict)
        and not any(
            inbox.get(key) is True for key in ("authorized", "configured", "enabled")
        )
    )
    capacity = policy.get("max_enabled_customers")
    if isinstance(capacity, bool) or not isinstance(capacity, int):
        raise TypeError("capacity")
    return candidate_matches, inbox_off, capacity, weekly.get("enabled") is True


def _receipt_health() -> tuple[bool, bool]:
    phase = _OBJECT.validate_json((EXECUTION / "phase.json").read_bytes())
    consumed = _OBJECT.validate_json(
        (AUTHORIZATION / "authorization-consumed.json").read_bytes()
    )
    credential = (
        (RUNTIME / "runtime-authority/task26-candidate-digest")
        .read_text(encoding="utf-8")
        .strip()
    )
    committed = (
        phase.get("phase") == "COMMITTED"
        and consumed.get("status") == "CONSUMED"
        and consumed.get("outcome") == "SUCCEEDED"
    )
    return committed, credential == CANDIDATE


def _telegram_sockets(pid: int) -> int:
    del pid
    count = 0
    for line in Path("/proc/net/tcp").read_text(encoding="ascii").splitlines()[1:]:
        fields = line.split()
        remote_address, remote_port = fields[2].split(":")
        address = ipaddress.ip_address(bytes.fromhex(remote_address)[::-1])
        if (
            fields[3] == "01"
            and int(remote_port, 16) == 443
            and any(address in network for network in _TELEGRAM_NETWORKS)
        ):
            count += 1
    return count


def _day_status_days(profile: Path = PROFILE) -> frozenset[date]:
    document = _OBJECT.validate_python(
        yaml.safe_load((profile / "config.yaml").read_text(encoding="utf-8"))
    )
    platforms = _mapping(document.get("platforms"), "platforms")
    telegram = _mapping(platforms.get("telegram"), "telegram")
    extra = _mapping(telegram.get("extra"), "extra")
    nutrition = _mapping(extra.get("nutrition_coaching"), "nutrition")
    configured = nutrition.get("weekly_operations_authority_path")
    if not isinstance(configured, str):
        raise TypeError("weekly_operations_authority_path")
    authority = profile / configured
    days: set[date] = set()
    for path in authority.glob("*.day-status-v1.jsonl"):
        for line in path.read_bytes().splitlines():
            row = _OBJECT.validate_json(line)
            raw = row.get("kst_day")
            if not isinstance(raw, str):
                raise TypeError("kst_day")
            days.add(date.fromisoformat(raw))
    return frozenset(days)


def collect() -> RuntimeSnapshot:
    """Collect one read-only snapshot from live production surfaces."""
    service = _systemd()
    config_candidate, inbox_off, capacity, weekly_enabled = _config_health()
    committed, credential_matches = _receipt_health()
    pid = int(service["MainPID"])
    journal = _run((
        "journalctl",
        "--user",
        "-u",
        SERVICE,
        "--since",
        "20 minutes ago",
        "--no-pager",
        "-o",
        "cat",
    ))
    return RuntimeSnapshot(
        active=service["ActiveState"] == "active"
        and service["SubState"] == "running"
        and pid > 0,
        candidate_matches=config_candidate
        and credential_matches
        and str(RUNTIME / "venv") in service["ExecStart"],
        channel_inbox_off=inbox_off,
        committed=committed,
        capacity=capacity,
        journal_clean=not any(fragment in journal for fragment in _BAD_JOURNAL),
        nrestarts=int(service["NRestarts"]),
        telegram_sockets=_telegram_sockets(pid),
        weekly_enabled=weekly_enabled,
        day_status_days=_day_status_days(),
    )


def append(
    observation: Observation,
    snapshot: RuntimeSnapshot,
    now: datetime,
) -> None:
    """Append one hash-linked private observation receipt."""
    OUTPUT.mkdir(parents=True, exist_ok=True, mode=0o700)
    lock_path = OUTPUT / "observer.lock"
    with lock_path.open("a+b") as lock:
        os.chmod(lock_path, 0o600)
        fcntl.flock(lock.fileno(), fcntl.LOCK_EX)
        log_path = OUTPUT / "observations.jsonl"
        previous = "0" * 64
        if log_path.is_file():
            rows = log_path.read_bytes().splitlines()
            if rows:
                last = _OBJECT.validate_json(rows[-1])
                digest = last.get("row_digest")
                if not isinstance(digest, str):
                    raise TypeError("row_digest")
                previous = digest
        snapshot_document: dict[str, JsonValue] = {
            "active": snapshot.active,
            "candidate_matches": snapshot.candidate_matches,
            "channel_inbox_off": snapshot.channel_inbox_off,
            "committed": snapshot.committed,
            "capacity": snapshot.capacity,
            "journal_clean": snapshot.journal_clean,
            "nrestarts": snapshot.nrestarts,
            "telegram_sockets": snapshot.telegram_sockets,
            "weekly_enabled": snapshot.weekly_enabled,
            "day_status_days": [
                day.isoformat() for day in sorted(snapshot.day_status_days)
            ],
        }
        body: dict[str, JsonValue] = {
            "schema": "nutricoach-v15-seven-day-observer-v1",
            "observed_at": now.isoformat(),
            "window_start": WINDOW_START.isoformat(),
            "window_end": WINDOW_END.isoformat(),
            "phase": observation.phase.value,
            "status": "PASS" if observation.passed else "FAIL",
            "failures": list(observation.failures),
            "previous_row_digest": previous,
            "snapshot": snapshot_document,
        }
        payload = json.dumps(body, sort_keys=True, separators=(",", ":")).encode()
        body["row_digest"] = hashlib.sha256(payload).hexdigest()
        encoded = json.dumps(body, sort_keys=True, separators=(",", ":")) + "\n"
        descriptor = os.open(
            log_path,
            os.O_APPEND | os.O_CREAT | os.O_WRONLY | os.O_CLOEXEC,
            0o600,
        )
        try:
            _ = os.write(descriptor, encoded.encode())
            os.fsync(descriptor)
        finally:
            os.close(descriptor)