"""Strict, read-only proof that a customer is absent from every staff chat."""

from __future__ import annotations

import fcntl
import hashlib
import json
import os
import stat
import uuid
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from enum import Enum
from pathlib import Path
from typing import Protocol


EVIDENCE_SCHEMA = "telegram-staff-membership-evidence-v1"
JOURNAL_SCHEMA = "telegram-staff-membership-journal-v1"
ABSENT_STATUSES = frozenset({"left", "kicked"})
ADMIN_STATUSES = frozenset({"administrator", "creator"})
DEFAULT_MAX_AGE = timedelta(minutes=5)


class MembershipGateError(ValueError):
    """Membership evidence is absent, stale, contradictory, or unverifiable."""


class StaffChatKind(str, Enum):
    MEMBERSHIP_CHAT = "membership_chat"
    PRIVATE_STAFF_DM = "private_staff_dm"


class ReadOnlyMembershipBot(Protocol):
    async def get_me(self) -> object: ...

    async def get_chat_member(self, chat_id: int, user_id: int) -> object: ...

    async def send_chat_action(
        self, chat_id: int, action: str, *, message_thread_id: int
    ) -> object: ...


@dataclass(frozen=True, slots=True)
class StaffChat:
    chat_id: int
    kind: StaffChatKind
    staff_user_id: int | None
    topic_ids: tuple[int, ...]
    roles: tuple[str, ...]
    sources: tuple[str, ...]

    def to_dict(self) -> dict[str, object]:
        result: dict[str, object] = {
            "chat_id": str(self.chat_id),
            "kind": self.kind.value,
            "topic_ids": list(self.topic_ids),
            "roles": list(self.roles),
            "sources": list(self.sources),
        }
        if self.staff_user_id is not None:
            result["staff_user_id"] = str(self.staff_user_id)
        return result


@dataclass(frozen=True, slots=True)
class StaffChatInventory:
    rows: tuple[StaffChat, ...]
    sha256: str

    def to_list(self) -> list[dict[str, object]]:
        return [row.to_dict() for row in self.rows]


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


def sha256_json(value: object) -> str:
    return hashlib.sha256(canonical_json(value)).hexdigest()


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


def _mapping(value: object, label: str) -> Mapping[str, object]:
    if not isinstance(value, Mapping) or not all(
        isinstance(key, str) for key in value
    ):
        raise MembershipGateError(f"{label} is invalid")
    return value


def _integer(value: object, label: str) -> int:
    if isinstance(value, bool) or not isinstance(value, (int, str)):
        raise MembershipGateError(f"{label} is invalid")
    try:
        result = int(value)
    except (TypeError, ValueError) as exc:
        raise MembershipGateError(f"{label} is invalid") from exc
    if str(result) != str(value).strip() or result == 0:
        raise MembershipGateError(f"{label} is invalid")
    return result


def _telegram_extra(config: Mapping[str, object]) -> Mapping[str, object]:
    platforms = config.get("platforms")
    if isinstance(platforms, Mapping):
        telegram = platforms.get("telegram")
        if isinstance(telegram, Mapping):
            extra = telegram.get("extra", telegram)
            return _mapping(extra, "Telegram configuration")
    telegram = config.get("telegram")
    if isinstance(telegram, Mapping):
        return _mapping(telegram.get("extra", telegram), "Telegram configuration")
    return config


@dataclass(slots=True)
class _InventoryAccumulator:
    chat_id: int
    kind: StaffChatKind
    staff_user_id: int | None
    topic_ids: set[int]
    roles: set[str]
    sources: set[str]


def build_staff_chat_inventory(
    registry: Mapping[str, object], config: Mapping[str, object]
) -> StaffChatInventory:
    """Enumerate and canonicalize every staff-accessible Telegram route."""
    rows: dict[int, _InventoryAccumulator] = {}

    def add(
        chat_value: object,
        *,
        source: str,
        role: str,
        user_value: object | None = None,
        topic_value: object | None = None,
        private_if_positive: bool = True,
    ) -> None:
        chat_id = _integer(chat_value, source)
        kind = (
            StaffChatKind.PRIVATE_STAFF_DM
            if chat_id > 0 and private_if_positive
            else StaffChatKind.MEMBERSHIP_CHAT
        )
        staff_user_id: int | None = None
        if kind is StaffChatKind.PRIVATE_STAFF_DM:
            staff_user_id = (
                chat_id if user_value is None else _integer(user_value, f"{source}.user_id")
            )
            if staff_user_id != chat_id:
                raise MembershipGateError(f"{source} private DM identity is contradictory")
        topic_id = (
            None if topic_value is None else _integer(topic_value, f"{source}.topic_id")
        )
        existing = rows.get(chat_id)
        if existing is not None:
            if existing.kind is not kind or existing.staff_user_id != staff_user_id:
                raise MembershipGateError(f"{source} staff route is contradictory")
            if topic_id is not None:
                existing.topic_ids.add(topic_id)
            existing.roles.add(role)
            existing.sources.add(source)
            return
        rows[chat_id] = _InventoryAccumulator(
            chat_id,
            kind,
            staff_user_id,
            set() if topic_id is None else {topic_id},
            {role},
            {source},
        )

    owner = _mapping(registry.get("owner"), "registry.owner")
    add(
        owner.get("chat_id"),
        source="registry.owner",
        role="owner",
        user_value=owner.get("user_id"),
    )
    extra = _telegram_extra(_mapping(config, "configuration"))
    nutrition = extra.get("nutrition_coaching")
    if nutrition is not None:
        nutrition_map = _mapping(nutrition, "nutrition_coaching")
        for key, role in (
            ("operator_review", "operator_review"),
            ("operator_card_route", "operator_card"),
        ):
            route = nutrition_map.get(key)
            if route is not None:
                route_map = _mapping(route, f"nutrition_coaching.{key}")
                add(
                    route_map.get("chat_id"),
                    source=f"platforms.telegram.extra.nutrition_coaching.{key}",
                    role=role,
                    user_value=route_map.get("user_id"),
                )
    adaptive = extra.get("adaptive_nutrition")
    if adaptive is not None:
        adaptive_map = _mapping(adaptive, "adaptive_nutrition")
        route = adaptive_map.get("review_operator")
        if route is not None:
            route_map = _mapping(route, "adaptive_nutrition.review_operator")
            add(
                route_map.get("chat_id"),
                source="platforms.telegram.extra.adaptive_nutrition.review_operator",
                role="adaptive_review",
                user_value=route_map.get("user_id"),
            )
        legacy_chat = adaptive_map.get("operator_chat_id")
        if legacy_chat is not None:
            add(
                legacy_chat,
                source="platforms.telegram.extra.adaptive_nutrition.operator_chat_id",
                role="adaptive_review",
                private_if_positive=False,
            )
    for key, role in (
        ("group_allowed_chats", "gateway_group"),
        ("allowed_chats", "gateway_allowed"),
    ):
        values = extra.get(key, ())
        if isinstance(values, str):
            values = tuple(item.strip() for item in values.split(",") if item.strip())
        if not isinstance(values, Sequence) or isinstance(values, (bytes, bytearray)):
            raise MembershipGateError(f"{key} is invalid")
        for value in values:
            if _integer(value, f"platforms.telegram.extra.{key}") > 0:
                raise MembershipGateError(
                    f"platforms.telegram.extra.{key} has an unknown positive route"
                )
            add(
                value,
                source=f"platforms.telegram.extra.{key}",
                role=role,
                private_if_positive=True,
            )
    topics = extra.get("group_topics", ())
    if not isinstance(topics, Sequence) or isinstance(topics, (str, bytes, bytearray)):
        raise MembershipGateError("group_topics is invalid")
    for topic in topics:
        topic_map = _mapping(topic, "group_topics entry")
        if _integer(topic_map.get("chat_id"), "group_topics.chat_id") > 0:
            raise MembershipGateError("group_topics has a positive group route")
        add(
            topic_map.get("chat_id"),
            source="platforms.telegram.extra.group_topics",
            role="group_topic",
            topic_value=topic_map.get("topic_id"),
            private_if_positive=False,
        )
    canonical_rows = tuple(
        StaffChat(
            item.chat_id,
            item.kind,
            item.staff_user_id,
            tuple(sorted(item.topic_ids)),
            tuple(sorted(item.roles)),
            tuple(sorted(item.sources)),
        )
        for item in sorted(rows.values(), key=lambda item: item.chat_id)
    )
    if not canonical_rows:
        raise MembershipGateError("staff chat inventory is empty")
    encoded = [row.to_dict() for row in canonical_rows]
    return StaffChatInventory(canonical_rows, sha256_json(encoded))


async def verify_subscription_readiness(
    bot: ReadOnlyMembershipBot,
    inventory: StaffChatInventory,
) -> int:
    """Require the bot to observe all membership chats as an administrator."""
    try:
        identity = await bot.get_me()
        bot_user_id = getattr(identity, "id")
    except Exception as exc:
        raise MembershipGateError("bot identity query failed") from exc
    if type(bot_user_id) is not int or bot_user_id <= 0:
        raise MembershipGateError("bot identity query failed")
    for row in inventory.rows:
        if row.kind is not StaffChatKind.MEMBERSHIP_CHAT:
            continue
        try:
            status = str(getattr(await bot.get_chat_member(row.chat_id, bot_user_id), "status"))
        except Exception as exc:
            raise MembershipGateError("bot administrator readiness query failed") from exc
        if status not in ADMIN_STATUSES:
            raise MembershipGateError(
                "bot is not administrator in every staff membership chat"
            )
        for topic_id in row.topic_ids:
            try:
                await bot.send_chat_action(
                    row.chat_id,
                    "typing",
                    message_thread_id=topic_id,
                )
            except Exception as exc:
                raise MembershipGateError(
                    "bot cannot reach every configured staff topic"
                ) from exc
    return bot_user_id


async def observe_customer_absence(
    bot: ReadOnlyMembershipBot,
    inventory: StaffChatInventory,
    *,
    customer_user_id: int,
) -> dict[str, object]:
    """Query only Bot identity and chat membership, failing on every uncertainty."""
    if type(customer_user_id) is not int or customer_user_id <= 0:
        raise MembershipGateError("customer Telegram identity is invalid")
    bot_user_id = await verify_subscription_readiness(bot, inventory)
    membership_rows = tuple(
        row for row in inventory.rows if row.kind is StaffChatKind.MEMBERSHIP_CHAT
    )
    results: list[dict[str, str]] = []
    for row in membership_rows:
        try:
            status = str(
                getattr(
                    await bot.get_chat_member(row.chat_id, customer_user_id),
                    "status",
                )
            )
        except Exception as exc:
            raise MembershipGateError("customer membership query failed") from exc
        if status not in ABSENT_STATUSES:
            raise MembershipGateError("customer is not absent from every staff chat")
        results.append({"chat_id": str(row.chat_id), "status": status})
    private_results: list[dict[str, object]] = []
    for row in inventory.rows:
        if row.kind is not StaffChatKind.PRIVATE_STAFF_DM:
            continue
        separated = (
            row.chat_id > 0
            and row.staff_user_id == row.chat_id
            and customer_user_id != row.chat_id
        )
        if not separated:
            raise MembershipGateError("customer identity overlaps a private staff DM")
        private_results.append(
            {
                "chat_id": str(row.chat_id),
                "identity_separated": True,
                "staff_user_id": str(row.staff_user_id),
            }
        )
    return {
        "bot_user_id_sha256": hashlib.sha256(str(bot_user_id).encode()).hexdigest(),
        "membership_results": results,
        "private_dm_results": private_results,
    }


class MembershipJournal:
    """Append-only mode-0600 hash chain for sanitized subscription events."""

    def __init__(self, path: Path) -> None:
        self.path = Path(path)
        self._lock_path = self.path.with_suffix(self.path.suffix + ".lock")

    def append(self, payload: Mapping[str, object]) -> dict[str, object]:
        self.path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
        self.path.parent.chmod(0o700)
        flags = os.O_CREAT | os.O_RDWR
        nofollow = getattr(os, "O_NOFOLLOW", 0)
        lock_fd = os.open(self._lock_path, flags | nofollow, 0o600)
        try:
            os.fchmod(lock_fd, 0o600)
            fcntl.flock(lock_fd, fcntl.LOCK_EX)
            rows = self.verify()
            body = dict(payload)
            forbidden = {"token", "username", "name", "raw_response", "message"}
            if forbidden.intersection(body):
                raise MembershipGateError("membership journal contains private fields")
            body["schema"] = JOURNAL_SCHEMA
            body["previous_row_sha256"] = (
                rows[-1]["row_sha256"] if rows else "0" * 64
            )
            digest = sha256_json(body)
            body["row_sha256"] = digest
            encoded = canonical_json(body) + b"\n"
            descriptor = os.open(
                self.path,
                os.O_CREAT | os.O_APPEND | os.O_WRONLY | nofollow,
                0o600,
            )
            try:
                os.fchmod(descriptor, 0o600)
                os.write(descriptor, encoded)
                os.fsync(descriptor)
            finally:
                os.close(descriptor)
            return body
        finally:
            fcntl.flock(lock_fd, fcntl.LOCK_UN)
            os.close(lock_fd)

    def append_transition(
        self,
        *,
        update_id: int,
        chat_id: int,
        customer_user_id: int,
        old_status: str,
        new_status: str,
        event_date_utc: str,
        subscription_epoch_id: str,
    ) -> dict[str, object]:
        rows = self.verify()
        transitions = [row for row in rows if row.get("event") == "transition"]
        for row in transitions:
            if row.get("update_id") == update_id:
                expected = {
                    "chat_id": str(chat_id),
                    "customer_user_id": str(customer_user_id),
                    "old_status": old_status,
                    "new_status": new_status,
                    "subscription_epoch_id": subscription_epoch_id,
                }
                if all(row.get(key) == value for key, value in expected.items()):
                    return row
                raise MembershipGateError("duplicate membership update conflicts")
        if transitions:
            previous_update_id = transitions[-1].get("update_id")
            if type(previous_update_id) is not int:
                raise MembershipGateError("membership journal is invalid")
            if previous_update_id > update_id:
                raise MembershipGateError("membership update ID regressed")
        if new_status not in ABSENT_STATUSES:
            raise MembershipGateError("customer entered a staff membership chat")
        return self.append(
            {
                "event": "transition",
                "update_id": update_id,
                "chat_id": str(chat_id),
                "customer_user_id": str(customer_user_id),
                "old_status": old_status,
                "new_status": new_status,
                "event_date_utc": event_date_utc,
                "subscription_epoch_id": subscription_epoch_id,
            }
        )

    def verify(self) -> list[dict[str, object]]:
        if not self.path.exists():
            return []
        metadata = self.path.lstat()
        if (
            self.path.is_symlink()
            or not stat.S_ISREG(metadata.st_mode)
            or stat.S_IMODE(metadata.st_mode) != 0o600
            or metadata.st_nlink != 1
        ):
            raise MembershipGateError("membership journal is not private")
        try:
            rows = [json.loads(line) for line in self.path.read_text().splitlines()]
        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise MembershipGateError("membership journal is invalid") from exc
        previous = "0" * 64
        for row in rows:
            if not isinstance(row, dict) or row.get("schema") != JOURNAL_SCHEMA:
                raise MembershipGateError("membership journal is invalid")
            digest = row.get("row_sha256")
            body = {key: value for key, value in row.items() if key != "row_sha256"}
            if row.get("previous_row_sha256") != previous or digest != sha256_json(body):
                raise MembershipGateError("membership journal hash chain is invalid")
            previous = str(digest)
        return rows


def write_private_json(path: Path, payload: Mapping[str, object]) -> None:
    path.parent.mkdir(mode=0o700, parents=True, exist_ok=True)
    path.parent.chmod(0o700)
    temporary = path.with_name(f".{path.name}.{uuid.uuid4().hex}.tmp")
    flags = os.O_CREAT | os.O_EXCL | os.O_WRONLY | getattr(os, "O_NOFOLLOW", 0)
    descriptor = os.open(temporary, flags, 0o600)
    try:
        encoded = canonical_json(payload) + b"\n"
        os.write(descriptor, encoded)
        os.fsync(descriptor)
    finally:
        os.close(descriptor)
    os.replace(temporary, path)
    path.chmod(0o600)


async def create_pre_activation_evidence(
    bot: ReadOnlyMembershipBot,
    inventory: StaffChatInventory,
    *,
    output_path: Path,
    deployment_receipt_path: Path,
    registry_path: Path,
    config_path: Path,
    customer_id: str,
    customer_user_id: int,
    bootstrap_session_id: str,
    bootstrap_generation: int,
    subscription_epoch_id: str,
    subscription_armed_at_utc: str,
    phase: str = "pre_activation",
    now: datetime | None = None,
) -> dict[str, object]:
    """Perform a fresh query and persist its complete immutable bindings."""
    current = now or datetime.now(UTC)
    started = current.isoformat()
    try:
        deployment = json.loads(deployment_receipt_path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise MembershipGateError("deployment receipt is unavailable") from exc
    required_deployment = (
        "candidate_digest", "candidate_core_digest", "candidate_inventory_digest",
        "hermes_wheel_sha256", "profile_wheel_sha256",
    )
    if not isinstance(deployment, dict) or any(
        not isinstance(deployment.get(field), str) for field in required_deployment
    ):
        raise MembershipGateError("deployment receipt is invalid")
    product_binding = deployment.get(
        "candidate_product_binding_sha256",
    )
    if (
        product_binding is not None
        and not isinstance(product_binding, str)
    ):
        raise MembershipGateError("deployment receipt is invalid")
    result = await observe_customer_absence(
        bot, inventory, customer_user_id=customer_user_id
    )
    observed = current.isoformat()
    if phase not in {"pre_activation", "post_lifecycle"}:
        raise MembershipGateError("membership evidence phase is invalid")
    payload: dict[str, object] = {
        "schema": EVIDENCE_SCHEMA,
        "phase": phase,
        **{field: deployment[field] for field in required_deployment},
        "deployment_receipt_path": str(deployment_receipt_path.resolve()),
        "deployment_receipt_sha256": sha256_file(deployment_receipt_path),
        "customer_id": customer_id,
        "customer_user_id": str(customer_user_id),
        "bootstrap_session_id": bootstrap_session_id,
        "bootstrap_generation": bootstrap_generation,
        "registry_sha256": sha256_file(registry_path),
        "config_sha256": sha256_file(config_path),
        "staff_chat_inventory": inventory.to_list(),
        "staff_chat_inventory_sha256": inventory.sha256,
        "subscription_epoch_id": subscription_epoch_id,
        "subscription_armed_at_utc": subscription_armed_at_utc,
        "request_started_at_utc": started,
        "observed_at_utc": observed,
        "activation_journal_sha256": "0" * 64,
        **result,
    }
    if isinstance(product_binding, str):
        payload["candidate_product_binding_sha256"] = (
            product_binding
        )
    payload["evidence_sha256"] = sha256_json(payload)
    write_private_json(output_path, payload)
    return payload


def validate_membership_evidence(
    path: Path,
    *,
    expected: Mapping[str, object],
    now: datetime | None = None,
    max_age: timedelta = DEFAULT_MAX_AGE,
) -> dict[str, object]:
    """Validate a complete candidate- and customer-bound pre-activation proof."""
    try:
        metadata = path.lstat()
        payload = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
        raise MembershipGateError("membership evidence is unavailable") from exc
    if (
        path.is_symlink()
        or not stat.S_ISREG(metadata.st_mode)
        or stat.S_IMODE(metadata.st_mode) != 0o600
        or metadata.st_nlink != 1
        or not isinstance(payload, dict)
        or payload.get("schema") != EVIDENCE_SCHEMA
        or payload.get("phase") != "pre_activation"
    ):
        raise MembershipGateError("membership evidence is invalid")
    evidence_digest = payload.get("evidence_sha256")
    body = {key: value for key, value in payload.items() if key != "evidence_sha256"}
    if evidence_digest != sha256_json(body):
        raise MembershipGateError("membership evidence digest is invalid")
    required_bindings = (
        "candidate_digest",
        "candidate_core_digest",
        "candidate_inventory_digest",
        "hermes_wheel_sha256",
        "profile_wheel_sha256",
        "deployment_receipt_sha256",
        "customer_id",
        "customer_user_id",
        "bootstrap_session_id",
        "bootstrap_generation",
        "registry_sha256",
        "config_sha256",
        "staff_chat_inventory_sha256",
        "subscription_epoch_id",
        "activation_journal_sha256",
    )
    for key in required_bindings:
        if key not in payload or key not in expected or payload[key] != expected[key]:
            raise MembershipGateError(f"membership evidence {key} is stale")
    binding_key = "candidate_product_binding_sha256"
    if (
        binding_key in payload
        or binding_key in expected
    ) and payload.get(binding_key) != expected.get(binding_key):
        raise MembershipGateError(
            f"membership evidence {binding_key} is stale",
        )
    inventory = payload.get("staff_chat_inventory")
    if not isinstance(inventory, list) or sha256_json(inventory) != payload.get(
        "staff_chat_inventory_sha256"
    ):
        raise MembershipGateError("membership evidence inventory is invalid")
    group_ids = {
        row.get("chat_id")
        for row in inventory
        if isinstance(row, dict) and row.get("kind") == StaffChatKind.MEMBERSHIP_CHAT.value
    }
    private_ids = {
        row.get("chat_id")
        for row in inventory
        if isinstance(row, dict) and row.get("kind") == StaffChatKind.PRIVATE_STAFF_DM.value
    }
    memberships = payload.get("membership_results")
    private = payload.get("private_dm_results")
    if (
        not isinstance(memberships, list)
        or {row.get("chat_id") for row in memberships if isinstance(row, dict)} != group_ids
        or any(
            not isinstance(row, dict) or row.get("status") not in ABSENT_STATUSES
            for row in memberships
        )
        or not isinstance(private, list)
        or {row.get("chat_id") for row in private if isinstance(row, dict)} != private_ids
        or any(
            not isinstance(row, dict) or row.get("identity_separated") is not True
            for row in private
        )
    ):
        raise MembershipGateError("membership evidence absence results are incomplete")
    try:
        started = datetime.fromisoformat(str(payload["request_started_at_utc"]))
        observed = datetime.fromisoformat(str(payload["observed_at_utc"]))
        armed = datetime.fromisoformat(str(payload["subscription_armed_at_utc"]))
    except (KeyError, ValueError) as exc:
        raise MembershipGateError("membership evidence timestamps are invalid") from exc
    current = now or datetime.now(UTC)
    if any(value.tzinfo is None for value in (started, observed, armed)):
        raise MembershipGateError("membership evidence timestamps are invalid")
    if not armed <= started <= observed <= current or current - observed > max_age:
        raise MembershipGateError("membership evidence is stale")
    return payload


__all__ = [
    "ABSENT_STATUSES",
    "EVIDENCE_SCHEMA",
    "MembershipGateError",
    "MembershipJournal",
    "ReadOnlyMembershipBot",
    "StaffChat",
    "StaffChatInventory",
    "StaffChatKind",
    "build_staff_chat_inventory",
    "canonical_json",
    "create_pre_activation_evidence",
    "observe_customer_absence",
    "sha256_file",
    "sha256_json",
    "validate_membership_evidence",
    "verify_subscription_readiness",
    "write_private_json",
]