"""Capability authority for operator-notification recovery."""

from __future__ import annotations

import hashlib
import json
import os
import stat
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Final, cast

_SCHEMA: Final = "operator-notification-recovery-authority-v1"
_CAPABILITY: Final = "operator_notification_recovery"


@dataclass(frozen=True, slots=True)
class OperatorNotificationRecoveryCapability:
    candidate_epoch: str
    owner_user_id: int
    owner_chat_id: int
    owner_topic_id: int
    authority_seal_sha256: str


def load_operator_notification_recovery(
    extra: Mapping[str, object],
    *,
    candidate_digest: str,
    owner_user_id: int,
    owner_chat_id: int,
    owner_topic_id: int,
) -> OperatorNotificationRecoveryCapability | None:
    room = extra.get("room_bootstrap")
    if not isinstance(room, Mapping):
        return None
    room = _string_mapping(cast(Mapping[object, object], room))
    raw = room.get("operator_notification_recovery")
    if raw is None:
        return None
    if not isinstance(raw, Mapping):
        raise ValueError("operator notification recovery config is invalid")
    raw = _string_mapping(cast(Mapping[object, object], raw))
    if raw.get("enabled") is not True:
        if set(raw) != {"enabled"}:
            raise ValueError("disabled operator notification recovery is invalid")
        return None
    if set(raw) != {"enabled", "receipt_path", "receipt_sha256"}:
        raise ValueError("operator notification recovery config is invalid")
    expected_file_digest = _digest(
        raw.get("receipt_sha256"),
        "receipt SHA",
    )
    receipt_path = _authorized_receipt_path(raw.get("receipt_path"))
    receipt_bytes = receipt_path.read_bytes()
    if not _same_digest(
        hashlib.sha256(receipt_bytes).hexdigest(),
        expected_file_digest,
    ):
        raise ValueError("operator notification recovery receipt SHA mismatch")
    try:
        document = json.loads(receipt_bytes)
    except json.JSONDecodeError as exc:
        raise ValueError(
            "operator notification recovery receipt is invalid"
        ) from exc
    if not isinstance(document, Mapping):
        raise ValueError("operator notification recovery receipt is invalid")
    document = _string_mapping(cast(Mapping[object, object], document))
    if (
        set(document)
        != {
            "authorization_seal_sha256",
            "capability",
            "candidate_digest",
            "owner",
            "schema",
            "status",
        }
        or document.get("schema") != _SCHEMA
        or document.get("capability") != _CAPABILITY
        or document.get("status") != "AUTHORIZED"
    ):
        raise ValueError("operator notification recovery receipt is invalid")
    sealed_candidate = _digest(
        document.get("candidate_digest"),
        "candidate digest",
    )
    _digest(candidate_digest, "active candidate digest")
    if not _same_digest(sealed_candidate, candidate_digest):
        raise ValueError("operator notification recovery candidate is invalid")
    owner = document.get("owner")
    if not isinstance(owner, Mapping):
        raise ValueError("operator notification recovery owner is invalid")
    owner = _string_mapping(cast(Mapping[object, object], owner))
    if (
        set(owner) != {"chat_id", "topic_id", "user_id"}
        or any(
            type(owner.get(name)) is not int
            for name in ("chat_id", "topic_id", "user_id")
        )
        or (
            owner.get("user_id"),
            owner.get("chat_id"),
            owner.get("topic_id"),
        )
        != (owner_user_id, owner_chat_id, owner_topic_id)
    ):
        raise ValueError("operator notification recovery owner is invalid")
    expected_seal = _digest(
        document.get("authorization_seal_sha256"),
        "authorization seal",
    )
    core = {
        key: value
        for key, value in document.items()
        if key != "authorization_seal_sha256"
    }
    actual_seal = hashlib.sha256(_canonical(core)).hexdigest()
    if not _same_digest(actual_seal, expected_seal):
        raise ValueError("operator notification recovery seal is invalid")
    return OperatorNotificationRecoveryCapability(
        candidate_epoch=sealed_candidate,
        owner_user_id=owner_user_id,
        owner_chat_id=owner_chat_id,
        owner_topic_id=owner_topic_id,
        authority_seal_sha256=expected_seal,
    )


def _authorized_receipt_path(value: object) -> Path:
    if not isinstance(value, str) or not value:
        raise ValueError("operator notification recovery receipt path is invalid")
    path = Path(value)
    root = _credential_root()
    if not path.is_absolute():
        if path.name != value:
            raise ValueError(
                "operator notification recovery receipt path is invalid"
            )
        path = root / path
    if path.is_symlink():
        raise ValueError("operator notification recovery receipt path is invalid")
    resolved = path.resolve(strict=True)
    if resolved.parent != root:
        raise ValueError("operator notification recovery receipt root is invalid")
    metadata = resolved.stat()
    if (
        not stat.S_ISREG(metadata.st_mode)
        or metadata.st_mode & 0o077
        or metadata.st_uid != os.getuid()
    ):
        raise ValueError("operator notification recovery receipt mode is invalid")
    return resolved


def _credential_root() -> Path:
    pin_value = os.environ.get("TASK26_AUTHORITY_PIN")
    if pin_value:
        return Path(pin_value).resolve(strict=True).parent
    credential_directory = os.environ.get("CREDENTIALS_DIRECTORY")
    if credential_directory:
        root = Path(credential_directory)
        if root.is_symlink():
            raise ValueError("systemd credential directory is invalid")
        return root.resolve(strict=True)
    raise ValueError("Task26 authority credentials are unavailable")


def _digest(value: object, label: str) -> str:
    if (
        not isinstance(value, str)
        or len(value) != 64
        or set(value) - set("0123456789abcdef")
    ):
        raise ValueError(f"operator notification recovery {label} is invalid")
    return value


def _canonical(value: object) -> bytes:
    return json.dumps(
        value,
        sort_keys=True,
        separators=(",", ":"),
    ).encode()


def _same_digest(left: str, right: str) -> bool:
    import hmac

    return hmac.compare_digest(left, right)


def _string_mapping(value: Mapping[object, object]) -> dict[str, object]:
    result: dict[str, object] = {}
    for key, item in value.items():
        if not isinstance(key, str):
            raise ValueError("operator notification recovery mapping is invalid")
        result[key] = item
    return result


__all__ = [
    "OperatorNotificationRecoveryCapability",
    "load_operator_notification_recovery",
]