#!/usr/bin/env python3
"""Disposable real-socket Telegram authority-revocation QA helper."""
from __future__ import annotations

import asyncio
import hashlib
import json
import os
import shutil
import socket
import tempfile
from pathlib import Path
from types import SimpleNamespace
from typing import Literal

from aiohttp import web

from gateway.config import PlatformConfig
from gateway.platforms.task26_runtime_authority import (
    append_external_authority,
    canonical as authority_canonical,
    digest as authority_digest,
)
from gateway.platforms.telegram import TelegramAdapter

RuntimeMode = Literal["source", "installed"]


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


def transcript_sha256(transcript: dict[str, object]) -> str:
    return hashlib.sha256(canonical(transcript)).hexdigest()


async def run_local_socket_telegram_qa(
    *, candidate_digest: str, runtime_mode: RuntimeMode
) -> dict[str, object]:
    profile = Path(tempfile.mkdtemp(prefix="task26-local-socket-profile-"))
    authority_root = Path(tempfile.mkdtemp(prefix="task26-local-socket-authority-"))
    profile.chmod(0o700)
    authority_root.chmod(0o700)
    registry = profile / "customers/registry.json"
    registry.parent.mkdir(mode=0o700)
    registry.write_bytes(
        canonical({"owner": {"user_id": "100", "chat_id": "100"}}) + b"\n"
    )
    registry.chmod(0o600)
    source_id = f"task26-{runtime_mode}-local-socket-authority"
    append_external_authority(
        authority_root,
        source_id=source_id,
        candidate_digest=hashlib.sha256(b"historical-local-socket").hexdigest(),
        action="qualify",
        historical_pass_digest=hashlib.sha256(b"historical-pass").hexdigest(),
        reason="historical local socket authority",
    )
    state = append_external_authority(
        authority_root,
        source_id=source_id,
        candidate_digest=candidate_digest,
        action="qualify",
        historical_pass_digest=hashlib.sha256(b"current-pass").hexdigest(),
        reason="current local socket authority",
    )
    pin: dict[str, object] = {
        "schema": "task26-authority-pin-v1",
        "authority_root": str(authority_root.resolve()),
        "source_id": source_id,
        "genesis_sha256": state["genesis_sha256"],
        "registry_head_sha256": state["registry_head_sha256"],
        "ledger_head_sha256": state["ledger_head_sha256"],
        "event_count": 2,
    }
    pin["pin_sha256"] = authority_digest(pin)
    pin_path = authority_root.parent / f".{authority_root.name}.pin.json"
    candidate_path = authority_root.parent / f".{authority_root.name}.candidate"
    pin_path.write_bytes(authority_canonical(pin) + b"\n")
    pin_path.chmod(0o600)
    candidate_path.write_text(candidate_digest + "\n", encoding="utf-8")
    candidate_path.chmod(0o600)

    request_seen = asyncio.Event()
    polling_live = asyncio.Event()
    polling_stopped = asyncio.Event()
    release_poll = asyncio.Event()
    authority_failed = asyncio.Event()
    requests: list[str] = []
    peer_ports: set[int] = set()
    active_connections = 0
    max_active_connections = 0
    inbound_updates_after_revocation = 0
    revocation_committed = False

    async def api(request: web.Request) -> web.Response:
        nonlocal active_connections, max_active_connections
        active_connections += 1
        max_active_connections = max(max_active_connections, active_connections)
        try:
            method = request.path.rsplit("/", 1)[-1]
            requests.append(method)
            peer = (
                request.transport.get_extra_info("peername")
                if request.transport is not None
                else None
            )
            if isinstance(peer, tuple) and len(peer) >= 2:
                peer_ports.add(int(peer[1]))
            request_seen.set()
            if method == "getMe":
                return web.json_response(
                    {
                        "ok": True,
                        "result": {
                            "id": 123456,
                            "is_bot": True,
                            "first_name": "Task26 Local",
                            "username": "task26_local_bot",
                        },
                    }
                )
            if method == "getUpdates":
                polling_live.set()
                if revocation_committed:
                    polling_stopped.set()
                    return web.json_response({"ok": True, "result": []})
                try:
                    await release_poll.wait()
                finally:
                    polling_stopped.set()
                return web.json_response({"ok": True, "result": []})
            return web.json_response({"ok": True, "result": True})
        finally:
            active_connections -= 1

    application = web.Application()
    application.router.add_post("/bot123456:task26/{method}", api)
    runner = web.AppRunner(application)
    await runner.setup()
    server_socket = socket.socket()
    server_socket.bind(("127.0.0.1", 0))
    port = server_socket.getsockname()[1]
    site = web.SockSite(runner, server_socket)
    await site.start()

    old_environment = {
        name: os.environ.get(name)
        for name in (
            "TASK26_AUTHORITY_PIN",
            "TASK26_CANDIDATE_DIGEST_FILE",
            "CREDENTIALS_DIRECTORY",
            "HERMES_TELEGRAM_DISABLE_FALLBACK_IPS",
            "TELEGRAM_PROXY",
            "HTTPS_PROXY",
            "HTTP_PROXY",
        )
    }
    os.environ["TASK26_AUTHORITY_PIN"] = str(pin_path)
    os.environ["TASK26_CANDIDATE_DIGEST_FILE"] = str(candidate_path)
    os.environ.pop("CREDENTIALS_DIRECTORY", None)
    os.environ["HERMES_TELEGRAM_DISABLE_FALLBACK_IPS"] = "true"
    for name in ("TELEGRAM_PROXY", "HTTPS_PROXY", "HTTP_PROXY"):
        os.environ.pop(name, None)

    adapter = TelegramAdapter(
        PlatformConfig(
            enabled=True,
            token="123456:task26",
            extra={
                "base_url": f"http://127.0.0.1:{port}/bot",
                "base_file_url": f"http://127.0.0.1:{port}/file/bot",
                "fallback_ips": ["127.0.0.1"],
                "nutrition_coaching": {
                    "enabled": True,
                    "registry_path": "customers/registry.json",
                    "profile_root": str(profile),
                },
            },
        )
    )

    async def no_staff_membership() -> None:
        return None

    original_capture = adapter._capture_telegram_update_context

    async def count_inbound(update: object, context: object) -> None:
        nonlocal inbound_updates_after_revocation
        if revocation_committed:
            inbound_updates_after_revocation += 1
        await original_capture(update, context)

    setattr(adapter, "_get_nutrition_coaching", lambda: SimpleNamespace())
    setattr(adapter, "_preflight_nutrition_generation_provider", lambda: True)
    setattr(adapter, "_arm_staff_membership_subscription", no_staff_membership)
    setattr(adapter, "_capture_telegram_update_context", count_inbound)
    original_authority_failed = adapter._task26_authority_failed

    def capture_failure(reason: str) -> None:
        original_authority_failed(reason)
        authority_failed.set()
        release_poll.set()

    setattr(adapter, "_task26_authority_failed", capture_failure)
    watcher_resources: dict[str, object] = {}
    disconnect_pass = False
    server_cleanup = False
    socket_cleanup = False
    watcher_cleanup = False
    server_socket_closed = False
    try:
        connect_task = asyncio.create_task(adapter.connect())
        await asyncio.wait_for(request_seen.wait(), timeout=2)
        if await asyncio.wait_for(connect_task, timeout=2) is not True:
            raise RuntimeError("local socket Telegram connect failed")
        await asyncio.wait_for(polling_live.wait(), timeout=2)
        watcher_resources = dict(adapter._task26_authority_watcher_resources)
        await asyncio.wait_for(
            asyncio.to_thread(
                append_external_authority,
                authority_root,
                source_id=source_id,
                candidate_digest=candidate_digest,
                action="revoke",
                historical_pass_digest=hashlib.sha256(b"live-revoke-pass").hexdigest(),
                reason="local socket live revocation",
            ),
            timeout=2,
        )
        revocation_committed = True
        await asyncio.wait_for(authority_failed.wait(), timeout=2)
        await asyncio.wait_for(adapter._task26_authority_disconnect_task, timeout=2)
        await asyncio.wait_for(polling_stopped.wait(), timeout=2)
        disconnect_pass = (
            adapter._app is None
            and adapter._bot is None
            and adapter._nutrition_coaching is None
        )
        watcher_cleanup = adapter._task26_authority_watcher is None
    finally:
        release_poll.set()
        await adapter.disconnect()
        await runner.cleanup()
        server_cleanup = not runner.sites
        server_socket_closed = server_socket.fileno() == -1
        socket_cleanup = server_socket_closed and active_connections == 0
        for name, value in old_environment.items():
            if value is None:
                os.environ.pop(name, None)
            else:
                os.environ[name] = value
        for root in (profile, authority_root):
            if root.exists():
                for path in root.rglob("*"):
                    if not path.is_symlink():
                        path.chmod(0o700 if path.is_dir() else 0o600)
                shutil.rmtree(root)
        pin_path.unlink(missing_ok=True)
        candidate_path.unlink(missing_ok=True)

    transcript: dict[str, object] = {
        "schema": "task26-local-http-telegram-api-qa-v1",
        "runtime_mode": runtime_mode,
        "candidate_digest": candidate_digest,
        "actual_socket": True,
        "mock_network": False,
        "backend": "python-telegram-bot-httpx-polling",
        "local_endpoint": {
            "scheme": "http",
            "host": "127.0.0.1",
            "port_class": "ephemeral_loopback",
            "api_path": "/bot<redacted>/{method}",
        },
        "connection_count": len(peer_ports),
        "request_count": len(requests),
        "methods": requests,
        "getMe_observed": "getMe" in requests,
        "getUpdates_observed": "getUpdates" in requests,
        "revocation_committed": revocation_committed,
        "disconnect": "PASS" if disconnect_pass else "FAIL",
        "post_revoke_updates": inbound_updates_after_revocation,
        "server_cleanup": server_cleanup,
        "socket_cleanup": socket_cleanup,
        "server_socket_closed": server_socket_closed,
        "active_connections": active_connections,
        "max_active_connections": max_active_connections,
        "watcher_cleanup": watcher_cleanup,
        "watch_backend": watcher_resources.get("backend"),
        "watch_directory_resource_count": watcher_resources.get(
            "directory_resource_count"
        ),
        "watch_inotify_count": watcher_resources.get("inotify_watch_count"),
        "external_traffic": False,
    }
    return transcript