#!/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()
    server_delivered_update = asyncio.Event()
    watcher_failure_deferred = asyncio.Event()
    requests: list[str] = []
    peer_ports: set[int] = set()
    active_connections = 0
    max_active_connections = 0
    server_delivered_post_revoke_updates = 0
    handler_entries = 0
    mutations = 0
    revocation_committed = False
    post_revoke_response_sent = False

    async def api(request: web.Request) -> web.Response:
        nonlocal active_connections, max_active_connections
        nonlocal server_delivered_post_revoke_updates
        nonlocal post_revoke_response_sent
        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 not revocation_committed:
                    await release_poll.wait()
                polling_stopped.set()
                if revocation_committed and not post_revoke_response_sent:
                    post_revoke_response_sent = True
                    server_delivered_post_revoke_updates += 1
                    server_delivered_update.set()
                    return web.json_response(
                        {
                            "ok": True,
                            "result": [
                                {
                                    "update_id": 7001,
                                    "message": {
                                        "message_id": 8001,
                                        "date": 1_787_094_000,
                                        "chat": {
                                            "id": 100,
                                            "type": "private",
                                            "first_name": "Task26 QA",
                                        },
                                        "from": {
                                            "id": 100,
                                            "is_bot": False,
                                            "first_name": "Task26 QA",
                                        },
                                        "text": "post-revocation update",
                                    },
                                }
                            ],
                        }
                    )
                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 handler_entries, mutations
        if revocation_committed:
            handler_entries += 1
            mutations += 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()

    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)

        def defer_watcher_disconnect(_reason: str) -> None:
            watcher_failure_deferred.set()

        watcher = adapter._task26_authority_watcher
        if watcher is None:
            raise RuntimeError("Task26 authority watcher is unavailable")
        setattr(watcher, "on_failure", defer_watcher_disconnect)
        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
        release_poll.set()
        await asyncio.wait_for(server_delivered_update.wait(), timeout=2)
        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": handler_entries,
        "server_delivered_post_revoke_updates": (
            server_delivered_post_revoke_updates
        ),
        "handler_entries": handler_entries,
        "mutations": mutations,
        "watcher_failure_deferred": watcher_failure_deferred.is_set(),
        "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