"""Candidate, permission, and clean-baseline contract for first-customer launch."""

from __future__ import annotations

import hashlib
import json
import os
import stat
import subprocess
from pathlib import Path
from typing import Final

from gateway.platforms import telegram_customer_bootstrap as bootstrap
from pydantic import TypeAdapter


CANDIDATE: Final = "0e383539aea1b83205771772f1e8b417840defe60f5ea6ce184f80e3af8d25f9"
HERMES_WHEEL_SHA256: Final = "ec160d3d0e29bc747f463923d31bfe736aefb5840a3d32fcb04842aa5139ddb0"
PROFILE_WHEEL_SHA256: Final = "a56da2417df0912f3fe407c0b78befd7362207d1c8a35271000af46701ba79d2"
TAGGED_SEAL_SHA256: Final = "8f26a2929dd349967ecc39df3b750fdd84a2ded1dfd09ffe4ea27263746a7604"
MODULE_SHA256: Final = "777e8ed7932d7834f39a67c8f8001742df99746ea4a990a5cedbc003fa51ab23"
BOT_USERNAME: Final = "nutricoach_kr_bot"
OWNER_ID: Final = "8693203710"
RELEASE: Final = Path(
    "/home/cube/projects/richard/traning coach/.omo/evidence/releases/dualcoach-v1.1.0"
)
HERMES_WHEEL: Final = RELEASE / "artifacts/hermes_agent-0.17.0-py3-none-any.whl"
PROFILE_WHEEL: Final = RELEASE / "artifacts/physique_checkin_cli-0.1.0-py3-none-any.whl"
TAGGED_SEAL: Final = RELEASE / "tagged-seal.json"
FORBIDDEN_ROOTS: Final = (
    "data/onboarding/telegram-customer-bootstrap-v1",
    "data/onboarding/telegram-publication-outbox-v1",
    "data/onboarding/telegram-staff-membership-v1",
)
type JsonValue = (
    None
    | bool
    | int
    | float
    | str
    | list["JsonValue"]
    | dict[str, "JsonValue"]
)
JSON_OBJECT: Final = TypeAdapter(dict[str, JsonValue])


class ControllerError(RuntimeError):
    """Fail-closed launch controller error."""


def sha256_file(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def json_object(path: Path) -> dict[str, JsonValue]:
    return JSON_OBJECT.validate_json(path.read_bytes())


def require_private_file(path: Path) -> None:
    info = path.lstat()
    if (
        path.is_symlink()
        or not stat.S_ISREG(info.st_mode)
        or stat.S_IMODE(info.st_mode) != 0o600
        or info.st_nlink != 1
        or info.st_uid != os.getuid()
    ):
        raise ControllerError(f"private file contract failed: {path}")


def exclusive_json(path: Path, value: dict[str, object]) -> None:
    if (
        path.parent.is_symlink()
        or not path.parent.is_dir()
        or stat.S_IMODE(path.parent.stat().st_mode) != 0o700
    ):
        raise ControllerError("output directory must be an existing private directory")
    raw = (
        json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":"))
        + "\n"
    ).encode()
    descriptor = os.open(
        path,
        os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_CLOEXEC | os.O_NOFOLLOW,
        0o600,
    )
    try:
        os.fchmod(descriptor, 0o600)
        os.write(descriptor, raw)
        os.fsync(descriptor)
    finally:
        os.close(descriptor)


def require_service_stopped(service: str) -> None:
    result = subprocess.run(
        [
            "systemctl",
            "--user",
            "show",
            service,
            "-p",
            "MainPID",
            "-p",
            "ActiveState",
            "-p",
            "SubState",
        ],
        check=False,
        capture_output=True,
        text=True,
    )
    values = dict(
        line.split("=", 1) for line in result.stdout.splitlines() if "=" in line
    )
    if result.returncode != 0 or values != {
        "MainPID": "0",
        "ActiveState": "inactive",
        "SubState": "dead",
    }:
        raise ControllerError("gateway must be inactive/dead with MainPID 0")


def clean_baseline(profile: Path, draft: Path) -> bootstrap.CustomerDraft:
    if (
        profile.is_symlink()
        or stat.S_IMODE(profile.stat().st_mode) != 0o700
        or stat.S_IMODE((profile / "data").stat().st_mode) != 0o700
    ):
        raise ControllerError("profile mode contract failed")
    registry = json_object(profile / "customers/registry.json")
    customers = registry.get("customers")
    if not isinstance(customers, list):
        raise ControllerError("registry customers are malformed")
    for customer in customers:
        if not isinstance(customer, dict):
            raise ControllerError("registry customer is malformed")
        consent = customer.get("ai_processing_consent")
        if customer.get("enabled") is True or (
            isinstance(consent, dict) and consent.get("granted") is True
        ):
            raise ControllerError("an enabled or consenting customer already exists")
    data_customers = profile / "data/customers"
    if data_customers.exists() and any(data_customers.iterdir()):
        raise ControllerError("customer operational data is not empty")
    if any((profile / relative).exists() for relative in FORBIDDEN_ROOTS):
        raise ControllerError("onboarding authority already exists")
    require_private_file(draft)
    parsed = bootstrap.load_customer_draft(draft)
    if parsed.customer_user_id is not None:
        raise ControllerError("draft customer identity must be unbound")
    return parsed