import asyncio
import tempfile
from pathlib import Path
from types import SimpleNamespace
from typing import Any

import pytest
from aiogram.exceptions import TelegramBadRequest, TelegramNetworkError
from aiogram.methods import SendMessage
from aiogram.types import Update
from claude_agent_sdk import PermissionResultAllow, PermissionResultDeny

from beaver_gateway.core.registry import McpRegistry
from beaver_gateway.core.transcript import build_entries
from beaver_gateway.frontends.base import GatewayRuntime
from beaver_gateway.frontends.telegram import TelegramFrontend
from beaver_gateway.frontends.telegram.render import chunks, status_label, to_html
from beaver_gateway.storage.models import ConversationBinding, Delivery, TelegramUpdate
from sqlmodel import select
from test_conversations import ScriptedClient, StubFrontend, World

USER = 42
GENERAL = f"{USER}/901"


class FakeBot:
    def __init__(self) -> None:
        self.updates: list[dict[str, Any]] = []
        self.sent: list[dict[str, Any]] = []
        self.drafts: list[dict[str, Any]] = []
        self.edits: list[dict[str, Any]] = []
        self.topics: list[str] = []
        self.reactions: list[tuple[int, list[Any]]] = []
        self.fail_sends = 0
        self.reject_html = False
        self._message_id = 100
        self.session = SimpleNamespace(close=self._close)

    async def _close(self) -> None:
        pass

    async def get_me(self) -> Any:
        return SimpleNamespace(username="bot", has_topics_enabled=True)

    async def get_updates(self, offset=None, **_: Any) -> list[Update]:
        await asyncio.sleep(0.01)
        return [
            Update.model_validate(u)
            for u in self.updates
            if offset is None or u["update_id"] >= offset
        ]

    async def send_message(
        self, chat_id, text, message_thread_id=None, parse_mode=None, **kwargs: Any
    ) -> Any:
        if self.fail_sends:
            self.fail_sends -= 1
            raise TelegramNetworkError(
                method=SendMessage(chat_id=0, text="x"), message="boom"
            )
        if self.reject_html and parse_mode == "HTML":
            raise TelegramBadRequest(
                method=SendMessage(chat_id=0, text="x"),
                message="Bad Request: can't parse entities",
            )
        self._message_id += 1
        self.sent.append(
            {
                "chat_id": chat_id,
                "thread": message_thread_id,
                "text": text,
                "parse_mode": parse_mode,
                "markup": kwargs.get("reply_markup"),
                "message_id": self._message_id,
            }
        )
        return SimpleNamespace(message_id=self._message_id)

    async def send_message_draft(self, **kwargs: Any) -> bool:
        self.drafts.append(kwargs)
        return True

    async def create_forum_topic(self, chat_id, name, **_: Any) -> Any:
        self.topics.append(name)
        return SimpleNamespace(message_thread_id=900 + len(self.topics), name=name)

    async def edit_forum_topic(
        self, chat_id, message_thread_id, name=None, **_: Any
    ) -> bool:
        self.topics.append(f"edit:{message_thread_id}:{name}")
        return True

    async def set_message_reaction(
        self, chat_id, message_id, reaction=None, **_: Any
    ) -> bool:
        self.reactions.append((message_id, reaction or []))
        return True

    async def answer_callback_query(self, *_: Any, **__: Any) -> bool:
        return True

    async def edit_message_text(self, text, chat_id, message_id, **_: Any) -> Any:
        self.edits.append({"message_id": message_id, "text": text})
        return True

    async def edit_message_reply_markup(
        self, chat_id, message_id, reply_markup=None
    ) -> Any:
        self.edits.append({"message_id": message_id, "markup": reply_markup})
        return True

    async def download(self, file_id, destination=None) -> None:
        Path(destination).write_bytes(b"data")

    # helpers for tests
    def push(self, payload: dict[str, Any]) -> None:
        payload["update_id"] = 1000 + len(self.updates)
        self.updates.append(payload)

    def message(self, text: str, *, thread: int | None = None, uid: int = USER) -> None:
        body: dict[str, Any] = {
            "message_id": 10 + len(self.updates),
            "date": 1700000000,
            "chat": {"id": USER, "type": "private"},
            "from": {"id": uid, "is_bot": False, "first_name": "h"},
            "text": text,
        }
        if thread is not None:
            body |= {"message_thread_id": thread, "is_topic_message": True}
        self.push({"message": body})

    def topic_created(self, name: str, thread: int) -> None:
        self.push(
            {
                "message": {
                    "message_id": 10 + len(self.updates),
                    "date": 1700000000,
                    "chat": {"id": USER, "type": "private"},
                    "from": {"id": USER, "is_bot": False, "first_name": "h"},
                    "message_thread_id": thread,
                    "is_topic_message": True,
                    "forum_topic_created": {"name": name, "icon_color": 1},
                }
            }
        )

    def callback(self, data: str, message_id: int) -> None:
        self.push(
            {
                "callback_query": {
                    "id": f"cb{len(self.updates)}",
                    "from": {"id": USER, "is_bot": False, "first_name": "h"},
                    "chat_instance": "ci",
                    "data": data,
                    "message": {
                        "message_id": message_id,
                        "date": 1700000000,
                        "chat": {"id": USER, "type": "private"},
                        "text": "?",
                    },
                }
            }
        )


class Stack:
    def __init__(self, world: World, bot: FakeBot) -> None:
        self.world = world
        self.bot = bot
        self.tg = TelegramFrontend(
            token="t",
            user_id=USER,
            master_agent="a",
            branch_agent="a",
            draft_interval=0.05,
            outbox_backoff=0.01,
        )
        self.tg._bot = bot  # noqa: SLF001
        markdown = StubFrontend("markdown", ("deep",), {"deep": "d"}, home=True)
        markdown.conversations = world.conversations
        frontends = [self.tg, markdown]
        world.conversations._frontends = frontends  # noqa: SLF001
        runtime = GatewayRuntime(
            agents=world.conversations._agents,  # noqa: SLF001
            mcps=McpRegistry([]),
            backends=world.conversations._backends,  # noqa: SLF001
            token_store=None,
            db=world.db,
            conversations=world.conversations,
            bus=world.bus,
            pool=world.pool,
            frontends=tuple(frontends),
        )
        self.tg.configure(runtime)
        self.tasks = [
            asyncio.create_task(self.tg.inbox.run()),
            asyncio.create_task(self.tg.outbox.run()),
            asyncio.create_task(self.tg._events()),  # noqa: SLF001
        ]

    async def close(self) -> None:
        for t in self.tasks:
            t.cancel()
        for t in self.tasks:
            with pytest.raises(BaseException):
                await t

    async def until(self, pred, timeout: float = 5.0, what: str = "condition") -> Any:
        deadline = asyncio.get_running_loop().time() + timeout
        while asyncio.get_running_loop().time() < deadline:
            value = pred()
            if value:
                return value
            await asyncio.sleep(0.02)
        msg = f"{what} never happened; sent={self.bot.sent}"
        raise AssertionError(msg)

    def sent_with(self, needle: str) -> dict[str, Any] | None:
        return next((m for m in self.bot.sent if needle in m["text"]), None)


@pytest.fixture
async def stack() -> Stack:
    root = Path(tempfile.mkdtemp(prefix="beaver-tg-"))
    world = await World(root).setup()
    await world.conversations.start()
    s = Stack(world, FakeBot())
    yield s
    await s.close()
    await world.conversations.stop()
    await world.pool.close_all()
    await world.db.dispose()


async def test_general_is_master_and_reply_has_no_thread(stack: Stack) -> None:
    stack.bot.message("hi")
    reply = await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    assert reply["thread"] == 901
    assert reply["parse_mode"] == "HTML"
    assert stack.bot.topics == ["🦫 General"]
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    assert master is not None and master.kind == "master"
    assert stack.bot.drafts and stack.bot.drafts[0]["chat_id"] == USER
    rows = await stack.world.conversations.queue.recent(master.id)
    assert {r.origin for r in rows} == {"сид:clean", "telegram"}


async def test_new_topic_becomes_morning_branch_and_replies_in_thread(
    stack: Stack,
) -> None:
    stack.bot.topic_created("план", 7)
    stack.bot.message("hello topic", thread=7)
    reply = await stack.until(lambda: stack.sent_with("ok:hello topic"), what="reply")
    assert reply["thread"] == 7
    branch = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/7"
    )
    assert branch is not None
    assert branch.kind == "branch" and branch.title == "план"
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    assert branch.parent_id == master.id
    prompts = [p for c in ScriptedClient.instances for p in c.prompts]
    seed = next(p for p in prompts if p.startswith("[сид: morning] branch «план»"))
    assert "Хендаут не приехал." in seed
    assert stack.bot.drafts[-1]["message_thread_id"] == 7


async def test_first_message_without_service_message_seeds_with_text(
    stack: Stack,
) -> None:
    stack.bot.message("сразу текстом", thread=8)
    await stack.until(lambda: stack.sent_with("сразу текстом"), what="reply")
    branch = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/8"
    )
    assert branch.title == "сразу текстом"
    client = next(
        c for c in ScriptedClient.instances if c.prompts and "сразу" in c.prompts[0]
    )
    assert client.prompts[0].startswith("[сид: morning]")
    assert client.prompts[0].endswith("сразу текстом")
    assert len(client.prompts) == 1


async def test_message_into_merged_branch_rebinds_the_topic(stack: Stack) -> None:
    stack.bot.message("one", thread=9)
    await stack.until(lambda: stack.sent_with("ok:"), what="first reply")
    old = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/9"
    )
    await stack.world.conversations.set_status(old, "merged")
    stack.bot.message("two", thread=9)
    await stack.until(lambda: stack.sent_with("two"), what="second reply")
    new = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/9"
    )
    assert new.id != old.id and new.status == "open"
    async with stack.world.db.session() as session:
        rows = list(
            (
                await session.exec(
                    select(ConversationBinding).where(
                        ConversationBinding.external_id == f"{USER}/9"
                    )
                )
            ).all()
        )
    assert {(r.conversation_id, r.visible) for r in rows} == {
        (old.id, False),
        (new.id, True),
    }


async def test_reply_from_another_window_is_mirrored_with_marker(stack: Stack) -> None:
    stack.bot.message("hi")
    await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    await stack.world.conversations.post(master, "from panel", origin="user")
    await stack.until(lambda: stack.sent_with("ok:from panel"), what="mirrored reply")
    texts = [m["text"] for m in stack.bot.sent]
    marker = next(i for i, t in enumerate(texts) if t.startswith("📝 из панели:"))
    assert "from panel" in texts[marker]
    assert texts[marker + 1] == "ok:from panel"


async def test_say_is_delivered_and_inject_turns_are_silent(stack: Stack) -> None:
    stack.bot.message("hi")
    await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    before = len(stack.bot.sent)
    await stack.world.conversations.inject(
        master, "cron tick", urgency="urgent", origin="крон"
    )
    await stack.world.settle(master, 3)
    await asyncio.sleep(0.2)
    assert len(stack.bot.sent) == before
    await stack.world.conversations.say(master, "psst")
    await stack.until(lambda: stack.sent_with("psst"), what="say")


async def test_question_becomes_buttons_and_callback_answers(stack: Stack) -> None:
    stack.bot.message("hi")
    await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    payload = {
        "questions": [
            {
                "header": "Цвет",
                "question": "Какой цвет?",
                "options": [
                    {"label": "Синий", "description": "как небо"},
                    {"label": "Красный", "description": ""},
                ],
                "multiSelect": False,
            }
        ]
    }
    asking = asyncio.create_task(
        stack.world.conversations.ask(master.external_id, payload)
    )
    question = await stack.until(
        lambda: stack.sent_with("Какой цвет?"), what="question"
    )
    assert question["markup"] is not None
    labels = [b.text for row in question["markup"].inline_keyboard for b in row]
    assert labels == ["Синий", "Красный"]
    pending = stack.world.conversations.pending_question(master.external_id)
    assert pending is not None
    stack.bot.callback(f"q:{pending[0]}:0:1", question["message_id"])
    assert await asyncio.wait_for(asking, 5) == "Красный"
    assert stack.world.conversations.answer_text("Красный") == (
        "Пользователь ответил: Красный"
    )
    await stack.until(
        lambda: any("✅ Красный" in e.get("text", "") for e in stack.bot.edits),
        what="edit",
    )
    row = await stack.world.conversations.get(master.external_id)
    assert row.pending_question is False


async def test_question_timeout_renders_text_and_free_text_answers(
    stack: Stack,
) -> None:
    stack.bot.message("hi")
    await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    stack.world.conversations._question_timeout = 0.3  # noqa: SLF001
    payload = {"questions": [{"header": "Q", "question": "Сколько?", "options": []}]}
    result = await stack.world.conversations.ask(master.external_id, payload)
    assert result is None
    assert "не ответил" in stack.world.conversations.answer_text(None)
    await stack.until(
        lambda: any("время вышло" in e.get("text", "") for e in stack.bot.edits),
        what="timeout edit",
    )
    stack.world.conversations._question_timeout = 5.0  # noqa: SLF001
    asking = asyncio.create_task(
        stack.world.conversations.ask(master.external_id, payload)
    )
    await stack.until(
        lambda: stack.world.conversations.pending_question(master.external_id),
        what="pending",
    )
    stack.bot.message("семь")
    assert await asyncio.wait_for(asking, 5) == "семь"


async def test_commands_status_merge_and_new(stack: Stack) -> None:
    stack.bot.message("/status")
    status = await stack.until(lambda: stack.sent_with("master · open"), what="status")
    assert "пул:" in status["text"]
    stack.bot.message("work", thread=11)
    await stack.until(lambda: stack.sent_with("work"), what="branch reply")
    branch = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/11"
    )
    await stack.world.store.append(
        stack.world.key(branch.session_id),
        build_entries(
            [{"role": "user", "content": "q"}, {"role": "assistant", "content": "a"}],
            session_id=branch.session_id,
            cwd=str(stack.world.root),
            model="m",
        ),
    )
    stack.bot.message("/merge", thread=11)
    await stack.until(lambda: stack.sent_with("слито в мастер"), what="merged")
    branch = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/11"
    )
    assert branch.status == "merged"
    stack.bot.message("/new отчёт")
    await stack.until(lambda: stack.sent_with("в новом топике"), what="new topic")
    assert stack.bot.topics == ["🦫 General", "отчёт"]
    child = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=f"{USER}/902"
    )
    assert child is not None and child.title == "отчёт"
    assert await stack.tg.mark_topic(child)
    assert stack.bot.topics[-1] == "edit:902:✅ отчёт"


async def test_master_topic_survives_rotation(stack: Stack) -> None:
    stack.bot.message("hi")
    await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    master = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    stack.bot.message("into general", thread=901)
    await stack.until(lambda: stack.sent_with("ok:into general"), what="reply")
    assert (await stack.world.conversations.find(kind="branch")) == []
    await stack.world.conversations.set_status(master, "closed")
    stack.bot.message("after rotation", thread=901)
    await stack.until(lambda: stack.sent_with("ok:after rotation"), what="reply")
    fresh = await stack.world.conversations.find_bound(
        frontend="telegram", external_id=GENERAL
    )
    assert fresh.kind == "master" and fresh.id != master.id
    assert stack.bot.topics == ["🦫 General"]
    assert (await stack.world.conversations.find(kind="branch")) == []


async def test_inbox_stores_first_and_replays_after_restart(stack: Stack) -> None:
    stack.bot.message("hi")
    await stack.until(lambda: stack.sent_with("ok:hi"), what="reply")
    async with stack.world.db.session() as session:
        rows = list((await session.exec(select(TelegramUpdate))).all())
    assert [r.update_id for r in rows] == [1000]
    assert rows[0].processed_at is not None and rows[0].error is None
    async with stack.world.db.session() as session:
        session.add(
            TelegramUpdate(
                update_id=999,
                payload={
                    "update_id": 999,
                    "message": {
                        "message_id": 1,
                        "date": 1700000000,
                        "chat": {"id": USER, "type": "private"},
                        "from": {"id": USER, "is_bot": False, "first_name": "h"},
                        "text": "replayed",
                    },
                },
            )
        )
        await session.commit()
    await stack.until(
        lambda: stack.sent_with("ok:replayed"), timeout=8, what="replayed reply"
    )
    stack.bot.message("stranger", uid=1)
    await asyncio.sleep(0.3)
    assert stack.sent_with("stranger") is None


async def test_outbox_retries_and_falls_back_to_plain(stack: Stack) -> None:
    stack.bot.fail_sends = 2
    stack.bot.reject_html = True
    stack.bot.message("**bold**")
    reply = await stack.until(lambda: stack.sent_with("ok:**bold**"), what="reply")
    assert reply["parse_mode"] is None
    async with stack.world.db.session() as session:
        rows = list((await session.exec(select(Delivery))).all())
    sent = [r for r in rows if r.status == "sent"]
    assert sent and sent[0].plain and sent[0].attempts >= 3


async def test_can_use_tool_answers_through_deny(stack: Stack) -> None:
    seen: list[tuple[str, dict[str, Any]]] = []

    async def asker(key: str, payload: dict[str, Any]) -> str:
        seen.append((key, payload))
        return "Пользователь ответил: да"

    backend = stack.world.backend
    backend._asker = asker  # noqa: SLF001
    can_use_tool = backend._can_use_tool("conv-1")  # noqa: SLF001
    allow = await can_use_tool("Bash", {"command": "ls"}, None)
    assert isinstance(allow, PermissionResultAllow)
    deny = await can_use_tool("AskUserQuestion", {"questions": []}, None)
    assert isinstance(deny, PermissionResultDeny)
    assert deny.message == "Пользователь ответил: да"
    assert seen == [("conv-1", {"questions": []})]


def test_render_helpers() -> None:
    assert (
        to_html("**жирно** и `code` <b>")
        == "<b>жирно</b> и <code>code</code> &lt;b&gt;"
    )
    assert to_html("# Заголовок\n- пункт") == "<b>Заголовок</b>\n• пункт"
    assert to_html("```py\nx = 1\n```") == "<pre>x = 1</pre>"
    assert (
        to_html("[док](https://a.b/c?x=1&y=2)")
        == '<a href="https://a.b/c?x=1&amp;y=2">док</a>'
    )
    parts = chunks("абв\n\n" + "г" * 30 + "\n\n" + "д" * 30, limit=40)
    assert parts == ["абв\n\n" + "г" * 30, "д" * 30]
    assert status_label("Read", {"file_path": "/x"}) == "читаю vault…"
    assert (
        status_label("mcp__firefly__list_accounts", None) == "firefly: list_accounts…"
    )
    assert status_label("Foo", {"command": "ls -la"}) == "Foo ls -la…"