from __future__ import annotations

import asyncio
from dataclasses import dataclass, field
from datetime import timedelta
import os
from pathlib import Path
from types import SimpleNamespace
from collections.abc import Callable, Sequence
from typing import Protocol, TypedDict
from unittest.mock import AsyncMock

import pytest
from telegram import Bot, Update
from telegram.ext import Updater

from gateway.platforms.telegram import _acknowledge_callback_best_effort
from gateway.platforms.telegram_polling_receipts import (
    ReceiptGatedTelegramBot,
    TelegramBusinessRecoveryCandidate,
    TelegramIngressReceiptError,
    TelegramIngressReceiptStore,
    TelegramPollingReceiptGate,
    set_current_telegram_update,
)


class _Update(Update):
    """Minimal real PTB update accepted by the polling seam."""


class _GetUpdatesOptions(TypedDict, total=False):
    offset: int | None
    timeout: timedelta


class _CallbackAnswerQuery(Protocol):
    async def answer(self, text: str | None = None) -> object: ...


@dataclass
class _AckQuery:
    id: str
    answer_mock: AsyncMock

    async def answer(self, text: str | None = None) -> object:
        if text is None:
            return await self.answer_mock()
        return await self.answer_mock(text=text)


@dataclass
class _OffsetAckServer:
    update: _Update
    pending: bool = True
    offsets: list[int] = field(default_factory=list)
    acknowledged: asyncio.Event = field(default_factory=asyncio.Event)
    empty_poll: asyncio.Event = field(default_factory=asyncio.Event)
    release_empty_poll: asyncio.Event = field(default_factory=asyncio.Event)

    async def get_updates(
        self,
        *,
        offset: int | None = None,
        timeout: timedelta | None = None,
        **_kwargs: object,
    ) -> tuple[_Update, ...]:
        requested_offset = offset if offset is not None else 0
        self.offsets.append(requested_offset)
        if self.pending and requested_offset > self.update.update_id:
            self.pending = False
            self.acknowledged.set()
            return ()
        if self.pending and requested_offset <= self.update.update_id:
            return (self.update,)
        if timeout == timedelta(0):
            return ()
        self.empty_poll.set()
        await self.release_empty_poll.wait()
        return ()


class _PollingBot(Bot):
    def __init__(self, server: _OffsetAckServer) -> None:
        super().__init__("123456:receipt-test-token")
        self._server = server
        self._initialized = False
        self._delete_webhook_calls: list[bool | None] = []

    @property
    def delete_webhook_calls(self) -> list[bool | None]:
        return self._delete_webhook_calls

    async def initialize(self) -> None:
        self._initialized = True

    async def delete_webhook(
        self,
        drop_pending_updates: bool | None = None,
        *,
        read_timeout: object = None,
        write_timeout: object = None,
        connect_timeout: object = None,
        pool_timeout: object = None,
        api_kwargs: object = None,
    ) -> bool:
        self._delete_webhook_calls.append(drop_pending_updates)
        return True

    async def get_updates(
        self,
        offset: int | None = None,
        limit: int | None = None,
        timeout: int | timedelta | None = None,
        allowed_updates: Sequence[str] | None = None,
        *,
        read_timeout: object = None,
        write_timeout: object = None,
        connect_timeout: object = None,
        pool_timeout: object = None,
        api_kwargs: object = None,
    ) -> tuple[Update, ...]:
        options: _GetUpdatesOptions = {"offset": offset}
        if isinstance(timeout, timedelta):
            options["timeout"] = timeout
        return await self._server.get_updates(**options)


class _ReceiptGatedPollingBot(ReceiptGatedTelegramBot, _PollingBot):
    """Preserve PTB's Bot contract while exercising the receipt proxy."""

    def __init__(
        self,
        bot: _PollingBot,
        gate: TelegramPollingReceiptGate,
        *,
        is_running: Callable[[], bool],
    ) -> None:
        super().__init__(bot, gate, is_running=is_running)
        self._initialized = False
        self._delete_webhook_calls = bot.delete_webhook_calls


@dataclass
class _PollingRun:
    updater: Updater
    queue: asyncio.Queue[object]
    gate: TelegramPollingReceiptGate
    blocked: asyncio.Event
    blocked_reasons: list[str]
    bot: _PollingBot


class _WriteFailingReceiptStore(TelegramIngressReceiptStore):
    def record(self, update_id: int) -> None:
        raise TelegramIngressReceiptError("receipt_write_failed")


async def _start_polling(
    server: _OffsetAckServer,
    store: TelegramIngressReceiptStore,
) -> _PollingRun:
    queue: asyncio.Queue[object] = asyncio.Queue()
    bot = _PollingBot(server)
    updater = Updater(bot, queue)
    blocked = asyncio.Event()
    blocked_reasons: list[str] = []

    def on_blocked(_update_id: int, reason_code: str) -> None:
        blocked_reasons.append(reason_code)
        updater._running = False
        blocked.set()

    gate = TelegramPollingReceiptGate(store, on_blocked=on_blocked)
    updater.bot = _ReceiptGatedPollingBot(
        bot,
        gate,
        is_running=lambda: updater.running,
    )
    await updater.initialize()
    await updater.start_polling(
        poll_interval=0,
        drop_pending_updates=False,
    )
    return _PollingRun(updater, queue, gate, blocked, blocked_reasons, bot)


async def _stop_polling(run: _PollingRun) -> None:
    if run.updater.running:
        await run.updater.stop()
    else:
        await run.updater._stop_polling()


async def _next_update(run: _PollingRun) -> _Update:
    update = await asyncio.wait_for(run.queue.get(), timeout=1)
    assert isinstance(update, _Update)
    return update


def test_receipt_store_contains_only_update_ids_and_fixed_safe_metadata(
    tmp_path: Path,
) -> None:
    path = tmp_path / "receipts.json"
    store = TelegramIngressReceiptStore(path)

    store.record(513)

    assert path.read_text(encoding="utf-8") == (
        '{\n'
        '  "receipts": {\n'
        '    "513": {\n'
        '      "reason_code": "handled",\n'
        '      "stage": "receipt"\n'
        '    }\n'
        '  },\n'
        '  "version": 1\n'
        '}'
    )
    assert path.stat().st_mode & 0o777 == 0o600

    store.record(514)
    store.forget_before(514)

    assert not store.contains(513)
    assert store.contains(514)


def test_recovered_business_commit_receipt_is_terminal_and_provenance_bound(
    tmp_path: Path,
) -> None:
    path = tmp_path / "receipts.json"
    store = TelegramIngressReceiptStore(path)
    provenance = "a" * 64

    store.record_recovered(629525050, provenance)
    store.forget_before(629525051)

    assert store.contains(629525050)
    assert store.recovered_provenance(629525050) == provenance
    document = __import__("json").loads(path.read_text(encoding="utf-8"))
    assert document == {
        "version": 2,
        "receipts": {},
        "terminal_receipts": {
            "629525050": {
                "stage": "recovery_receipt",
                "reason_code": "business_commit_reconciled",
                "provenance_digest": provenance,
            }
        },
    }
    with pytest.raises(
        TelegramIngressReceiptError,
        match="^receipt_store_corrupt$",
    ):
        store.record_recovered(629525050, "b" * 64)


def test_business_recovery_does_not_overwrite_historical_handler_exception(
    tmp_path: Path,
) -> None:
    store = TelegramIngressReceiptStore(tmp_path / "receipts.json")
    gate = TelegramPollingReceiptGate(store, on_blocked=lambda *_args: None)
    update = _Update(629525050)
    gate.captured([update])
    gate.failed(update, "handler_exception")

    with pytest.raises(TelegramIngressReceiptError, match="^handler_exception$"):
        gate.reconcile_business_commit(
            update_id=629525050,
            provenance_digest="a" * 64,
        )

    assert store.recovered_provenance(629525050) is None
    assert store.failure_reason(629525050) == "handler_exception"

    restarted = TelegramPollingReceiptGate(store, on_blocked=lambda *_args: None)
    with pytest.raises(TelegramIngressReceiptError, match="^handler_exception$"):
        restarted.reconcile_business_commit(
            update_id=629525050,
            provenance_digest="a" * 64,
        )
    assert store.recovered_provenance(629525050) is None


@pytest.mark.asyncio
async def test_authenticated_callback_replay_promotes_registered_business_recovery(
    tmp_path: Path,
) -> None:
    store = TelegramIngressReceiptStore(tmp_path / "receipts.json")
    gate = TelegramPollingReceiptGate(store, on_blocked=lambda *_args: None)
    candidate = TelegramBusinessRecoveryCandidate(
        update_id=700,
        provenance_digest="c" * 64,
        actor_id=12,
        chat_id=-100,
        topic_id=22,
        message_id=120,
        callback_data="signed-owner-callback",
    )
    gate.register_business_recovery(candidate)
    update = SimpleNamespace(
        update_id=700,
        callback_query=SimpleNamespace(
            data="signed-owner-callback",
            from_user=SimpleNamespace(id=12),
            message=SimpleNamespace(
                chat=SimpleNamespace(id=-100),
                message_thread_id=22,
                message_id=120,
            ),
        ),
    )

    assert await gate.begin(update) is True
    assert store.recovered_provenance(700) == "c" * 64


def _assert_receipt_path_symlink_fails_closed_without_touching_target(
    tmp_path: Path,
    target_contents: str | None,
) -> None:
    path = tmp_path / "receipts.json"
    target = tmp_path / "outside-authority.json"
    if target_contents is not None:
        target.write_text(target_contents, encoding="utf-8")
    path.symlink_to(target)

    with pytest.raises(
        TelegramIngressReceiptError,
        match="^receipt_store_corrupt$",
    ):
        TelegramIngressReceiptStore(path).record(601)

    assert path.is_symlink()
    if target_contents is None:
        assert not os.path.lexists(target)
    else:
        assert target.read_text(encoding="utf-8") == target_contents


@pytest.mark.parametrize(
    "target_contents",
    [None, "outside-authority"],
    ids=["dangling", "existing-target"],
)
def test_receipt_path_symlink_attacks_fail_closed(
    tmp_path: Path,
    target_contents: str | None,
) -> None:
    _assert_receipt_path_symlink_fails_closed_without_touching_target(
        tmp_path,
        target_contents,
    )


def test_receipt_path_hardlink_fifo_and_mode_fail_closed(
    tmp_path: Path,
) -> None:
    source = tmp_path / "source.json"
    source.write_text("{}", encoding="utf-8")
    os.chmod(source, 0o600)
    hardlink = tmp_path / "hardlink.json"
    os.link(source, hardlink)
    fifo = tmp_path / "receipt.fifo"
    os.mkfifo(fifo, 0o600)
    wrong_mode = tmp_path / "wrong-mode.json"
    wrong_mode.write_text("{}", encoding="utf-8")
    os.chmod(wrong_mode, 0o640)

    for path in (hardlink, fifo, wrong_mode):
        with pytest.raises(
            TelegramIngressReceiptError,
            match="^receipt_store_corrupt$",
        ):
            TelegramIngressReceiptStore(path).record(602)


def test_lock_path_symlink_attacks_fail_closed_without_touching_target(
    tmp_path: Path,
) -> None:
    for target_contents in (None, "outside-authority"):
        path = tmp_path / f"receipts-{target_contents is None}.json"
        target = tmp_path / f"outside-lock-{target_contents is None}.json"
        if target_contents is not None:
            target.write_text(target_contents, encoding="utf-8")
        lock_path = path.with_suffix(path.suffix + ".lock")
        lock_path.symlink_to(target)

        with pytest.raises(
            TelegramIngressReceiptError,
            match="^receipt_lock_failed$",
        ):
            TelegramIngressReceiptStore(path).record(603)

        assert lock_path.is_symlink()
        if target_contents is None:
            assert not os.path.lexists(target)
        else:
            assert target.read_text(encoding="utf-8") == target_contents


def test_receipt_writer_replaces_a_raced_symlink_without_following_target(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    path = tmp_path / "receipts.json"
    target = tmp_path / "outside-authority.json"
    replace = os.replace

    def replace_after_symlink(
        source: str | bytes | os.PathLike[str] | os.PathLike[bytes],
        destination: str | bytes | os.PathLike[str] | os.PathLike[bytes],
        *,
        src_dir_fd: int | None = None,
        dst_dir_fd: int | None = None,
    ) -> None:
        path.symlink_to(target)
        replace(
            source,
            destination,
            src_dir_fd=src_dir_fd,
            dst_dir_fd=dst_dir_fd,
        )

    monkeypatch.setattr(os, "replace", replace_after_symlink)
    store = TelegramIngressReceiptStore(path)
    store.record(604)

    assert not path.is_symlink()
    assert not os.path.lexists(target)
    assert store.contains(604)
    assert path.stat().st_mode & 0o777 == 0o600


@pytest.mark.asyncio
async def test_callback_ack_failure_uses_the_ptb_update_id_not_callback_data(
    caplog: pytest.LogCaptureFixture,
) -> None:
    set_current_telegram_update(_Update(514))
    query: _CallbackAnswerQuery = _AckQuery(
        id="callback-id-must-not-be-logged",
        answer_mock=AsyncMock(
            side_effect=RuntimeError("answer-must-not-be-logged")
        ),
    )

    with caplog.at_level("WARNING", logger="gateway.platforms.telegram"):
        await _acknowledge_callback_best_effort(
            query,
            gate="nutrition_customer_consent",
        )

    assert (
        "telegram_callback_ack_failed "
        "gate=nutrition_customer_consent update_id=514 error=RuntimeError"
        in caplog.text
    )
    assert "callback-id-must-not-be-logged" not in caplog.text
    assert "answer-must-not-be-logged" not in caplog.text


@pytest.mark.asyncio
async def test_real_ptb_offset_stays_unacknowledged_until_durable_receipt(
    tmp_path: Path,
) -> None:
    server = _OffsetAckServer(_Update(515))
    store = TelegramIngressReceiptStore(tmp_path / "receipts.json")
    first = await _start_polling(server, store)

    captured = await _next_update(first)
    assert captured.update_id == 515
    assert server.offsets == [0]
    assert first.bot.delete_webhook_calls == [False]
    assert not store.contains(515)

    await _stop_polling(first)
    second = await _start_polling(server, store)
    replayed = await _next_update(second)
    assert replayed.update_id == 515
    assert await second.gate.begin(replayed) is False
    second.gate.completed(replayed)
    await asyncio.wait_for(server.acknowledged.wait(), timeout=1)

    assert server.offsets == [0, 0, 516]
    assert not store.contains(515)
    await _stop_polling(second)


@pytest.mark.asyncio
async def test_real_ptb_replays_after_business_persistence_before_ingress_receipt(
    tmp_path: Path,
) -> None:
    server = _OffsetAckServer(_Update(516))
    store = TelegramIngressReceiptStore(tmp_path / "receipts.json")
    durable_business_updates: set[int] = set()

    first = await _start_polling(server, store)
    captured = await _next_update(first)
    assert await first.gate.begin(captured) is False
    durable_business_updates.add(captured.update_id)
    await _stop_polling(first)

    second = await _start_polling(server, store)
    replayed = await _next_update(second)
    assert await second.gate.begin(replayed) is False
    durable_business_updates.add(replayed.update_id)
    second.gate.completed(replayed)
    await asyncio.wait_for(server.acknowledged.wait(), timeout=1)

    assert durable_business_updates == {516}
    assert server.offsets == [0, 0, 517]
    await _stop_polling(second)


@pytest.mark.asyncio
async def test_real_ptb_duplicate_after_receipt_skips_business_before_offset_ack(
    tmp_path: Path,
) -> None:
    server = _OffsetAckServer(_Update(517))
    store = TelegramIngressReceiptStore(tmp_path / "receipts.json")
    business_runs: list[int] = []

    first = await _start_polling(server, store)
    captured = await _next_update(first)
    assert await first.gate.begin(captured) is False
    business_runs.append(captured.update_id)
    first.gate.completed(captured)
    await _stop_polling(first)
    assert server.offsets == [0]
    assert store.contains(517)

    second = await _start_polling(server, store)
    replayed = await _next_update(second)
    assert await second.gate.begin(replayed) is True
    await asyncio.wait_for(server.acknowledged.wait(), timeout=1)

    assert business_runs == [517]
    assert server.offsets == [0, 0, 518]
    assert not store.contains(517)
    await _stop_polling(second)


@pytest.mark.asyncio
async def test_real_ptb_handler_failure_blocks_the_offset_without_retrying(
    tmp_path: Path,
) -> None:
    server = _OffsetAckServer(_Update(518))
    run = await _start_polling(
        server,
        TelegramIngressReceiptStore(tmp_path / "receipts.json"),
    )
    captured = await _next_update(run)
    assert await run.gate.begin(captured) is False
    run.gate.failed(captured, "handler_exception")
    await asyncio.wait_for(run.blocked.wait(), timeout=1)

    assert run.blocked_reasons == ["handler_exception"]
    assert server.offsets == [0]
    await _stop_polling(run)


@pytest.mark.asyncio
async def test_real_ptb_receipt_write_failure_and_corruption_block_the_offset(
    tmp_path: Path,
) -> None:
    write_server = _OffsetAckServer(_Update(519))
    write_run = await _start_polling(
        write_server,
        _WriteFailingReceiptStore(tmp_path / "write-failure.json"),
    )
    write_update = await _next_update(write_run)
    assert await write_run.gate.begin(write_update) is False
    write_run.gate.completed(write_update)
    await asyncio.wait_for(write_run.blocked.wait(), timeout=1)
    assert write_run.blocked_reasons == ["receipt_write_failed"]
    assert write_server.offsets == [0]
    await _stop_polling(write_run)

    corrupt_path = tmp_path / "corrupt.json"
    corrupt_path.write_text("not-json", encoding="utf-8")
    corrupt_server = _OffsetAckServer(_Update(520))
    corrupt_run = await _start_polling(
        corrupt_server,
        TelegramIngressReceiptStore(corrupt_path),
    )
    corrupt_update = await _next_update(corrupt_run)
    assert await corrupt_run.gate.begin(corrupt_update) is True
    await asyncio.wait_for(corrupt_run.blocked.wait(), timeout=1)
    assert corrupt_run.blocked_reasons == ["receipt_store_corrupt"]
    assert corrupt_server.offsets == [0]
    await _stop_polling(corrupt_run)


@pytest.mark.asyncio
async def test_real_ptb_cancellation_before_receipt_replays_without_offset_ack(
    tmp_path: Path,
) -> None:
    server = _OffsetAckServer(_Update(521))
    store = TelegramIngressReceiptStore(tmp_path / "receipts.json")
    first = await _start_polling(server, store)
    business_updates: set[int] = set()
    started = asyncio.Event()
    release = asyncio.Event()

    async def cancelled_handler() -> None:
        update = await _next_update(first)
        assert await first.gate.begin(update) is False
        business_updates.add(update.update_id)
        started.set()
        await release.wait()
        first.gate.completed(update)

    task = asyncio.create_task(cancelled_handler())
    await asyncio.wait_for(started.wait(), timeout=1)
    task.cancel()
    with pytest.raises(asyncio.CancelledError):
        await task
    await _stop_polling(first)
    assert server.offsets == [0]
    assert not store.contains(521)

    second = await _start_polling(server, store)
    replayed = await _next_update(second)
    assert await second.gate.begin(replayed) is False
    business_updates.add(replayed.update_id)
    second.gate.completed(replayed)
    await asyncio.wait_for(server.acknowledged.wait(), timeout=1)

    assert business_updates == {521}
    assert server.offsets == [0, 0, 522]
    await _stop_polling(second)