import asyncio
from telethon import types

from app.linkswap import compute_message_changes
from app.replace_engine import ScanOptions, apply_channel, scan_channel, undo_batch
from app.tgutil import button_url, is_callback_button

from .dbutil import fresh_db, db_url_for
from .fakes import FakeClient, cb_btn, make_msg, markup, rpc, url_btn

OLD = "https://t.me/bharath?start=BBB"


def build():
    text3 = "Hi 😀 https://t.me/bharath?start=CCC done"
    msgs = {
        1: make_msg(1, "Get it now", [types.MessageEntityTextUrl(0, 3, "https://t.me/Bharath?start=AAA")]),
        2: make_msg(
            2,
            "Episodes",
            markup=markup([url_btn("Ep 1-20", OLD), cb_btn("👍 3", b"r1")], [url_btn("Join", "https://t.me/other")]),
            out=True,
        ),
        3: make_msg(3, text3, [types.MessageEntityBold(37, 4)]),
        4: make_msg(4, "see https://t.me/bharath_bot?start=1"),
        5: make_msg(5, "https://t.me/bharath/123"),
        7: types.MessageEmpty(id=7, peer_id=types.PeerChannel(1)),
    }
    return msgs, text3


async def setup(tmp_path, msgs, pts=8):
    db = await fresh_db(db_url_for(tmp_path))
    await db.save_channel(1, 5, "Test", None, 99)
    ch = await db.get_channel(1)
    return db, ch, FakeClient(msgs, pts=pts)


def test_scan_apply_undo(tmp_path):
    async def run():
        msgs, text3 = build()
        db, ch, client = await setup(tmp_path, msgs)
        opts = ScanOptions("bharath", "goku")
        scan = await scan_channel(client, ch, opts, can_edit_others=True)
        assert set(scan.infos) == {1, 2, 3}
        assert scan.skipped_post_links == 1  # t.me/bharath/123 is a channel post, not a bot link
        assert scan.foreign == 2 and scan.infos[2].mine and scan.infos[2].callbacks

        await db.create_batch("b1", "bharath", "goku", 99)
        res = await apply_channel(client, db, scan, opts, "b1", 99, edit_delay=0)
        assert (res.edited, res.failed) == (3, [])

        # hyperlink: only the username changed
        assert msgs[1].entities[0].url == "https://t.me/goku?start=AAA"
        # buttons: URL swapped, reaction button + other link untouched, text untouched
        rows = msgs[2].reply_markup.rows
        assert button_url(rows[0].buttons[0]) == "https://t.me/goku?start=BBB"
        assert is_callback_button(rows[0].buttons[1])
        assert button_url(rows[1].buttons[0]) == "https://t.me/other"
        assert msgs[2].message == "Episodes"
        # typed link after an emoji: text swapped and the following entity moved with it
        assert msgs[3].message == "Hi 😀 https://t.me/goku?start=CCC done"
        assert (msgs[3].entities[0].offset, msgs[3].entities[0].length) == (34, 4)
        # untouched
        assert msgs[4].message == "see https://t.me/bharath_bot?start=1"
        assert msgs[5].message == "https://t.me/bharath/123"

        # adopted into the database
        p = await db.find_by_message(1, 3)
        assert p.source == "adopted" and "goku" in p.text
        assert (await db.find_by_message(1, 2)).buttons[0][0]["u"].endswith("goku?start=BBB")

        # undo restores everything
        ures = await undo_batch(client, db, "b1", edit_delay=0)
        assert ures.restored == 3 and not ures.failed
        assert msgs[1].entities[0].url == "https://t.me/Bharath?start=AAA"
        assert msgs[3].message == text3 and msgs[3].entities[0].offset == 37
        assert button_url(msgs[2].reply_markup.rows[0].buttons[0]) == OLD
        assert is_callback_button(msgs[2].reply_markup.rows[0].buttons[1])
        assert (await db.last_batch()) is None  # batch marked as undone
        await db.close()

    asyncio.run(run())


def test_only_own_posts_without_edit_right(tmp_path):
    async def run():
        msgs, _ = build()
        db, ch, client = await setup(tmp_path, msgs)
        opts = ScanOptions("bharath", "goku")
        scan = await scan_channel(client, ch, opts, can_edit_others=False)
        assert set(scan.infos) == {2} and scan.not_editable == 2
        await db.create_batch("b2", "bharath", "goku", 99)
        res = await apply_channel(client, db, scan, opts, "b2", 99, edit_delay=0)
        assert res.edited == 1 and len(client.edits) == 1
        await db.close()

    asyncio.run(run())


def test_failures_are_reported_not_fatal(tmp_path):
    async def run():
        msgs, _ = build()
        db, ch, client = await setup(tmp_path, msgs)
        client.fail[1] = rpc("MessageAuthorRequiredError")
        opts = ScanOptions("bharath", "goku")
        scan = await scan_channel(client, ch, opts)
        await db.create_batch("b3", "bharath", "goku", 99)
        res = await apply_channel(client, db, scan, opts, "b3", 99, edit_delay=0)
        assert res.edited == 2 and res.failed == [(1, "MessageAuthorRequiredError")]
        await db.close()

    asyncio.run(run())


def test_scan_without_counter_falls_back_to_probing(tmp_path):
    async def run():
        msgs, _ = build()
        db, ch, client = await setup(tmp_path, msgs, pts=None)
        scan = await scan_channel(client, ch, ScanOptions("bharath", "goku"))
        assert set(scan.infos) == {1, 2, 3}
        await db.close()

    asyncio.run(run())


def test_post_links_can_be_included():
    m = make_msg(5, "https://t.me/bharath/123")
    assert compute_message_changes(m, "bharath", "goku")[0] is None
    ch, skipped = compute_message_changes(m, "bharath", "goku", include_posts=True)
    assert ch.new_text == "https://t.me/goku/123"


def test_tg_scheme_and_hyperlink_and_text_flags():
    m = make_msg(
        6,
        "open tg://resolve?domain=bharath&start=x",
        [types.MessageEntityTextUrl(0, 4, "tg://resolve?domain=Bharath&start=y")],
    )
    ch, _ = compute_message_changes(m, "bharath", "goku")
    assert ch.new_text == "open tg://resolve?domain=goku&start=x"
    assert ch.new_entities[0].url == "tg://resolve?domain=goku&start=y"
    assert ch.n_typed == 1 and ch.n_links == 1

    ch2, _ = compute_message_changes(m, "bharath", "goku", include_typed=False)
    assert ch2.new_text == m.message and ch2.n_typed == 0