"""Uploader tests: destination policy and direct streaming to Telegram."""

from __future__ import annotations

import asyncio
import io
from types import SimpleNamespace
from typing import Any

import pytest
from aiogram.exceptions import (
    TelegramBadRequest,
    TelegramForbiddenError,
    TelegramNetworkError,
    TelegramRetryAfter,
    TelegramServerError,
)
from aiogram.types import (
    BufferedInputFile,
    ChatMemberAdministrator,
    ChatMemberMember,
    ChatMemberOwner,
    User,
)
from PIL import Image

from app.database.models import Job, JobStatus, Video, VideoStatus
from app.errors import (
    ContentTooLargeError,
    DestinationError,
    InvalidMediaError,
    JobCancelledError,
    NetworkError,
    NotFoundError,
    UploadRejectedError,
    UploadUncertainError,
)
from app.services.uploader import HttpStreamInputFile, Uploader, normalize_chat_target
from tests.helpers import MP4_HEADER


class FakeBot:
    id = 4242

    def __init__(
        self,
        *,
        chat: Any = None,
        member: Any = None,
        chat_error: Exception | None = None,
        member_error: Exception | None = None,
    ) -> None:
        self.chat = chat
        self.member = member
        self.chat_error = chat_error
        self.member_error = member_error
        self.video_calls: list[dict[str, Any]] = []
        self.message_calls: list[dict[str, Any]] = []
        self.video_errors: list[BaseException] = []
        self.streamed: list[bytes] = []
        self.thumbnail_data: list[bytes | None] = []

    async def get_chat(self, target: Any) -> Any:
        if self.chat_error is not None:
            raise self.chat_error
        return self.chat

    async def get_chat_member(self, chat_id: Any, user_id: Any) -> Any:
        if self.member_error is not None:
            raise self.member_error
        return self.member

    async def send_video(self, **kwargs: Any) -> Any:
        self.video_calls.append(kwargs)
        if self.video_errors:
            raise self.video_errors.pop(0)
        video = kwargs["video"]
        assert isinstance(video, HttpStreamInputFile)
        body = bytearray()
        async for chunk in video.read(self):
            body.extend(chunk)
        self.streamed.append(bytes(body))
        thumbnail = kwargs.get("thumbnail")
        self.thumbnail_data.append(thumbnail.data if isinstance(thumbnail, BufferedInputFile) else None)
        return SimpleNamespace(message_id=100 + len(self.video_calls))

    async def send_message(self, **kwargs: Any) -> Any:
        self.message_calls.append(kwargs)
        return SimpleNamespace(message_id=500 + len(self.message_calls))


class FakeHttp:
    def __init__(
        self,
        payload: bytes | None = None,
        error: Exception | None = None,
        *,
        stream_error: Exception | None = None,
    ) -> None:
        self.payload = payload
        self.error = error
        self.stream_error = stream_error
        self.calls: list[tuple[str, int]] = []
        self.stream_calls: list[dict[str, Any]] = []

    async def fetch_bytes(
        self,
        url: str,
        *,
        max_bytes: int,
        allowed_prefixes: Any = (),
        referer: str | None = None,
    ) -> bytes:
        self.calls.append((url, max_bytes))
        if self.error is not None:
            raise self.error
        return self.payload or b""

    async def iter_media(
        self,
        url: str,
        *,
        max_bytes: int,
        cancel_event: asyncio.Event | None = None,
        referer: str | None = None,
    ):
        self.stream_calls.append(
            {
                "url": url,
                "max_bytes": max_bytes,
                "referer": referer,
                "cancel_event": cancel_event,
            }
        )
        if self.stream_error is not None:
            raise self.stream_error
        yield MP4_HEADER


def _member_owner() -> ChatMemberOwner:
    return ChatMemberOwner(
        status="creator",
        user=User(id=4242, is_bot=True, first_name="Bot"),
        is_anonymous=False,
    )


def _member_admin(*, can_post: bool | None) -> ChatMemberAdministrator:
    return ChatMemberAdministrator(
        user=User(id=4242, is_bot=True, first_name="Bot"),
        can_be_edited=False,
        is_anonymous=False,
        can_manage_chat=True,
        can_delete_messages=True,
        can_manage_video_chats=True,
        can_restrict_members=True,
        can_promote_members=True,
        can_change_info=True,
        can_invite_users=True,
        can_post_stories=False,
        can_edit_stories=False,
        can_delete_stories=False,
        can_send_welcome_messages=False,
        can_post_messages=can_post,
    )


def _member_plain() -> ChatMemberMember:
    return ChatMemberMember(user=User(id=4242, is_bot=True, first_name="Bot"))


def _chat(chat_type: str = "channel", chat_id: int = -100123) -> SimpleNamespace:
    return SimpleNamespace(id=chat_id, type=chat_type, title="Test destination")


async def _resolve(uploader, repository) -> Any:
    await repository.set_setting("destination_chat_id", "@my_channel")
    return await uploader.resolve_destination()


def _uploader(
    settings,
    repository,
    bot: FakeBot | None = None,
    http: FakeHttp | None = None,
) -> tuple[Uploader, FakeBot, FakeHttp]:
    bot = bot or FakeBot(chat=_chat(), member=_member_admin(can_post=True))
    http = http or FakeHttp()
    return Uploader(settings, bot, repository, http), bot, http


async def _stored_video(
    repository,
    *,
    status: VideoStatus = VideoStatus.DOWNLOADED,
    body: bytes = MP4_HEADER,
    caption: str = "یک ویدئوی آزمایشی",
    duration: float | None = 12.5,
    thumbnail_url: str | None = None,
    media_url: str | None = "https://example.com/video.mp4",
    file_size: int | None = None,
) -> Video:
    video = Video.create(
        job_id="a" * 32,
        source_url="https://example.com/video/1",
        page_url="https://example.com/video/1",
        title="عنوان آزمایشی",
        caption=caption,
        media_type="video/mp4",
        media_url=media_url,
        thumbnail_url=thumbnail_url,
        file_size=file_size,
    )
    if await repository.get_job("a" * 32) is None:
        await repository.create_job(
            Job(
                id="a" * 32,
                source_url="https://example.com/list",
                status=JobStatus.PENDING,
                owner_id=1,
                owner_chat_id=1,
                correlation_id="a" * 12,
            )
        )
    video.status = status
    video, _ = await repository.add_video(video)
    await repository.update_video_fields(
        video.id,
        local_path=None,
        file_size=len(body) if file_size is None else file_size,
        duration_seconds=duration,
    )
    stored = await repository.get_video(video.id)
    assert stored is not None
    return stored


def test_normalize_chat_target_accepts_usernames_and_numbers() -> None:
    assert normalize_chat_target("@my_channel") == "@my_channel"
    assert normalize_chat_target(" -1001234567890 ") == -1001234567890
    assert normalize_chat_target("123") == 123


@pytest.mark.parametrize("raw", ["", "   ", "0", "abc", "@ab", "@bad-name", "12.5", "@"])
def test_normalize_chat_target_rejects_invalid_values(raw: str) -> None:
    with pytest.raises(DestinationError):
        normalize_chat_target(raw)


@pytest.mark.asyncio
async def test_resolve_destination_requires_configuration(settings, repository) -> None:
    uploader, _, _ = _uploader(settings, repository)
    with pytest.raises(DestinationError, match="تنظیم نشده"):
        await uploader.resolve_destination()


@pytest.mark.asyncio
async def test_resolve_destination_reads_runtime_setting(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    destination = await _resolve(uploader, repository)
    assert destination.chat_id == -100123
    assert destination.chat_type == "channel"
    assert bot.chat is not None


@pytest.mark.asyncio
async def test_resolve_destination_malformed_setting_falls_back_to_env(
    settings, repository
) -> None:
    settings.default_chat_id = -100999
    uploader, _, _ = _uploader(settings, repository)
    await repository.set_setting("destination_chat_id", "not-valid")
    destination = await uploader.resolve_destination()
    assert destination.chat_id == -100123  # FakeBot still returns its configured chat


@pytest.mark.asyncio
async def test_private_chat_destination_is_rejected(settings, repository) -> None:
    bot = FakeBot(chat=_chat("private"), member=_member_plain())
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError, match="کانال یا گروه"):
        await uploader.validate_target(-100123)


@pytest.mark.asyncio
async def test_channel_requires_bot_administrator(settings, repository) -> None:
    bot = FakeBot(chat=_chat("channel"), member=_member_plain())
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError, match="مدیر"):
        await uploader.validate_target(-100123)


@pytest.mark.asyncio
async def test_channel_requires_post_permission(settings, repository) -> None:
    bot = FakeBot(chat=_chat("channel"), member=_member_admin(can_post=False))
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError, match="ارسال پیام"):
        await uploader.validate_target(-100123)


@pytest.mark.asyncio
async def test_channel_owner_and_admin_with_permission_are_accepted(
    settings, repository
) -> None:
    owner_bot = FakeBot(chat=_chat("channel"), member=_member_owner())
    uploader, _, _ = _uploader(settings, repository, bot=owner_bot)
    assert (await uploader.validate_target(-100123)).chat_type == "channel"

    admin_bot = FakeBot(chat=_chat("channel"), member=_member_admin(can_post=True))
    uploader, _, _ = _uploader(settings, repository, bot=admin_bot)
    assert (await uploader.validate_target(-100123)).chat_type == "channel"


@pytest.mark.asyncio
async def test_group_destination_checks_membership(settings, repository) -> None:
    bot = FakeBot(chat=_chat("supergroup"), member=_member_plain())
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    destination = await uploader.validate_target(-100123)
    assert destination.chat_type == "supergroup"

    kicked = SimpleNamespace(status="kicked", can_send_messages=True)
    bot = FakeBot(chat=_chat("group"), member=kicked)
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError, match="حذف"):
        await uploader.validate_target(-100123)


@pytest.mark.asyncio
async def test_destination_lookup_errors_are_wrapped(settings, repository) -> None:
    bot = FakeBot(
        chat_error=TelegramForbiddenError(None, "Forbidden: bot is not a member"),
    )
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError, match="دسترسی"):
        await uploader.validate_target(-100123)

    bot = FakeBot(chat=_chat(), member_error=TelegramBadRequest(None, "Bad Request"))
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError):
        await uploader.validate_target(-100123)

    bot = FakeBot(chat=_chat(), chat_error=TelegramBadRequest(None, "Bad Request: chat not found"))
    uploader, _, _ = _uploader(settings, repository, bot=bot)
    with pytest.raises(DestinationError, match="پیدا نشد"):
        await uploader.validate_target(-100123)


@pytest.mark.asyncio
async def test_send_test_message(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    destination = await uploader.send_test_message("@my_channel")
    assert destination.chat_type == "channel"
    assert bot.message_calls and "بررسی شد" in bot.message_calls[0]["text"]


@pytest.mark.asyncio
async def test_upload_streams_bytes_without_a_local_file(settings, repository) -> None:
    settings.send_thumbnails = False
    uploader, bot, http = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)

    result = await uploader.upload(video, destination, asyncio.Event())

    assert result.message_id == 101
    assert result.extra_messages == 0
    call = bot.video_calls[0]
    assert call["chat_id"] == -100123
    assert call["caption"] == "یک ویدئوی آزمایشی"
    assert call["parse_mode"] is None
    assert call["supports_streaming"] is True
    assert call["duration"] == 12
    assert call["thumbnail"] is None
    assert isinstance(call["video"], HttpStreamInputFile)
    assert call["video"].filename.endswith(".mp4")
    assert bot.streamed == [MP4_HEADER]
    assert http.stream_calls[0]["referer"] == "https://example.com/video/1"
    assert http.stream_calls[0]["max_bytes"] == min(
        settings.max_file_size_bytes, settings.telegram_upload_limit_bytes
    )


@pytest.mark.asyncio
async def test_upload_is_idempotent_for_already_uploaded_video(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository, status=VideoStatus.UPLOADED)
    await repository.update_video_fields(video.id, telegram_message_id=777)
    video = await repository.get_video(video.id)
    destination = await _resolve(uploader, repository)

    result = await uploader.upload(video, destination, asyncio.Event())
    assert result.message_id == 777
    assert bot.video_calls == []


@pytest.mark.asyncio
async def test_upload_with_unknown_status_is_never_resent(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository, status=VideoStatus.UPLOAD_UNKNOWN)
    destination = await _resolve(uploader, repository)
    with pytest.raises(UploadUncertainError):
        await uploader.upload(video, destination, asyncio.Event())
    assert bot.video_calls == []


@pytest.mark.asyncio
async def test_upload_rejects_missing_media_url(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    destination = await _resolve(uploader, repository)
    video = await _stored_video(repository, media_url=None)
    with pytest.raises(UploadRejectedError, match="لینک مستقیم"):
        await uploader.upload(video, destination, asyncio.Event())
    assert bot.video_calls == []


@pytest.mark.asyncio
async def test_upload_rejects_declared_oversized_file_before_streaming(
    settings, repository
) -> None:
    settings.telegram_max_upload_mb = 1
    settings.max_file_size_mb = 2
    uploader, bot, http = _uploader(settings, repository)
    destination = await _resolve(uploader, repository)
    video = await _stored_video(repository, file_size=1024 * 1024 + 100)
    with pytest.raises(UploadRejectedError, match="بیشتر"):
        await uploader.upload(video, destination, asyncio.Event())
    assert bot.video_calls == []
    assert http.stream_calls == []


@pytest.mark.asyncio
async def test_upload_honours_cancellation_before_send(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    event = asyncio.Event()
    event.set()
    with pytest.raises(JobCancelledError):
        await uploader.upload(video, destination, event)
    assert bot.video_calls == []


@pytest.mark.asyncio
async def test_upload_retries_flood_wait_once(settings, repository, monkeypatch) -> None:
    settings.retry_count = 1
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    bot.video_errors = [TelegramRetryAfter(None, "Too Many Requests", 1)]

    async def no_wait(seconds: float, cancel_event: asyncio.Event) -> bool:
        return False

    monkeypatch.setattr(Uploader, "_wait_or_cancel", staticmethod(no_wait))
    result = await uploader.upload(video, destination, asyncio.Event())
    assert len(bot.video_calls) == 2
    assert result.message_id == 102


@pytest.mark.asyncio
async def test_upload_rejects_long_flood_wait(settings, repository) -> None:
    settings.retry_count = 3
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    bot.video_errors = [TelegramRetryAfter(None, "Too Many Requests", 301)]
    with pytest.raises(UploadRejectedError, match="محدودیت نرخ"):
        await uploader.upload(video, destination, asyncio.Event())
    assert len(bot.video_calls) == 1


@pytest.mark.asyncio
async def test_upload_exhausted_flood_wait_is_rejected(settings, repository) -> None:
    settings.retry_count = 0
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    bot.video_errors = [TelegramRetryAfter(None, "Too Many Requests", 1)]
    with pytest.raises(UploadRejectedError, match="چند تلاش"):
        await uploader.upload(video, destination, asyncio.Event())


@pytest.mark.asyncio
@pytest.mark.parametrize(
    "error",
    [
        TelegramNetworkError(None, "connection reset"),
        TelegramServerError(None, "internal"),
    ],
)
async def test_network_and_server_errors_are_ambiguous_and_never_retried(
    settings, repository, error: Exception
) -> None:
    settings.retry_count = 3
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    bot.video_errors = [error]
    with pytest.raises(UploadUncertainError):
        await uploader.upload(video, destination, asyncio.Event())
    assert len(bot.video_calls) == 1


@pytest.mark.asyncio
async def test_telegram_rejections_become_upload_rejected(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)

    bot.video_errors = [TelegramBadRequest(None, "Bad Request: file is too big")]
    with pytest.raises(UploadRejectedError, match="بیش از حد بزرگ"):
        await uploader.upload(video, destination, asyncio.Event())

    bot.video_errors = [TelegramBadRequest(None, "Bad Request: chat not found")]
    with pytest.raises(UploadRejectedError):
        await uploader.upload(video, destination, asyncio.Event())


@pytest.mark.asyncio
async def test_unexpected_send_error_is_ambiguous(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    bot.video_errors = [RuntimeError("socket exploded")]
    with pytest.raises(UploadUncertainError):
        await uploader.upload(video, destination, asyncio.Event())
    assert len(bot.video_calls) == 1


@pytest.mark.asyncio
async def test_source_stream_errors_are_mapped_without_duplicate_sending(
    settings, repository
) -> None:
    uploader, bot, _ = _uploader(settings, repository, http=FakeHttp(stream_error=ContentTooLargeError()))
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)
    with pytest.raises(UploadRejectedError, match="سقف"):
        await uploader.upload(video, destination, asyncio.Event())

    uploader, bot, _ = _uploader(settings, repository, http=FakeHttp(stream_error=InvalidMediaError()))
    with pytest.raises(UploadRejectedError, match="معتبر"):
        await uploader.upload(video, destination, asyncio.Event())

    uploader, bot, _ = _uploader(settings, repository, http=FakeHttp(stream_error=NotFoundError()))
    with pytest.raises(UploadRejectedError, match="ناموفق"):
        await uploader.upload(video, destination, asyncio.Event())

    uploader, bot, _ = _uploader(settings, repository, http=FakeHttp(stream_error=NetworkError("cut")))
    with pytest.raises(UploadUncertainError):
        await uploader.upload(video, destination, asyncio.Event())
    assert len(bot.video_calls) == 1


@pytest.mark.asyncio
async def test_upload_without_message_id_is_ambiguous(settings, repository) -> None:
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository)
    destination = await _resolve(uploader, repository)

    async def no_id(**kwargs: Any) -> Any:
        bot.video_calls.append(kwargs)
        return SimpleNamespace(message_id=0)

    bot.send_video = no_id  # type: ignore[method-assign]
    with pytest.raises(UploadUncertainError, match="message ID"):
        await uploader.upload(video, destination, asyncio.Event())


@pytest.mark.asyncio
async def test_long_caption_is_split_into_extra_messages(settings, repository) -> None:
    settings.telegram_caption_limit = 50
    uploader, bot, _ = _uploader(settings, repository)
    video = await _stored_video(repository, caption="متن " * 80)
    destination = await _resolve(uploader, repository)

    result = await uploader.upload(video, destination, asyncio.Event())
    assert result.extra_messages >= 1
    assert len(bot.message_calls) == result.extra_messages
    assert all(call["parse_mode"] is None for call in bot.message_calls)
    assert all(call["disable_web_page_preview"] is True for call in bot.message_calls)


@pytest.mark.asyncio
async def test_remote_thumbnail_is_prepared_in_memory(settings, repository) -> None:
    settings.send_thumbnails = True
    image_bytes = io.BytesIO()
    Image.new("RGB", (640, 360), (30, 120, 200)).save(image_bytes, format="PNG")
    http = FakeHttp(payload=image_bytes.getvalue())
    uploader, bot, _ = _uploader(settings, repository, http=http)
    video = await _stored_video(repository, thumbnail_url="https://example.com/thumb.png")
    destination = await _resolve(uploader, repository)

    result = await uploader.upload(video, destination, asyncio.Event())
    assert result.message_id == 101
    assert http.calls and http.calls[0][0] == "https://example.com/thumb.png"
    thumbnail = bot.video_calls[0]["thumbnail"]
    assert isinstance(thumbnail, BufferedInputFile)
    assert bot.thumbnail_data[0] is not None
    with Image.open(io.BytesIO(bot.thumbnail_data[0] or b"")) as image:
        assert image.format == "JPEG"
        assert image.width <= 320
        assert len(bot.thumbnail_data[0] or b"") <= 200 * 1024


def test_thumbnail_processing_rejects_invalid_images() -> None:
    assert Uploader._thumbnail_bytes(b"not an image") is None


def test_thumbnail_processing_creates_small_jpeg() -> None:
    image_bytes = io.BytesIO()
    Image.new("RGB", (1280, 720), (200, 50, 50)).save(image_bytes, format="PNG")
    data = Uploader._thumbnail_bytes(image_bytes.getvalue())
    assert data is not None
    with Image.open(io.BytesIO(data)) as image:
        assert image.format == "JPEG"
        assert image.width <= 320
    assert len(data) <= 200 * 1024