"""The native Discord client walks what gallery-dl walked, in its order (#4412).

No network: a fake session answers by endpoint. What these pin is the part of
gallery-dl's behaviour that decides WHICH files exist and WHAT they are called —
the channel walk, the file list and its numbering, the name split — because a
difference there is a re-download or a missed file at cutover, not a style
choice.
"""

from __future__ import annotations

from datetime import date

import pytest

from backend.app.services import discord_client as dc
from backend.app.services.discord_client import (
    DiscordAPIError,
    DiscordAuthError,
    DiscordClient,
    firefox_user_agent,
    message_text,
    nameext_from_url,
    parse_source_url,
)


class _Resp:
    def __init__(self, status, body=None, headers=None):
        self.status_code = status
        self._body = body
        self.headers = headers or {}
        self.content = b"x"

    def json(self):
        return self._body


class _Session:
    """Answers `GET API_ROOT + endpoint` from a dict. A value may be a list of
    responses, served in order; a messages entry is keyed by (endpoint, before)."""

    def __init__(self, routes):
        self.routes = routes
        self.headers = {}
        self.calls = []

    def get(self, url, params=None, timeout=None):
        endpoint = url[len(dc.API_ROOT):]
        self.calls.append((endpoint, dict(params or {})))
        key = endpoint
        if endpoint.endswith("/messages"):
            key = (endpoint, (params or {}).get("before"))
        elif endpoint.endswith("/threads/search"):
            key = (endpoint, (params or {}).get("offset"))
        answer = self.routes.get(key, _Resp(404, {}))
        if isinstance(answer, list):
            return answer.pop(0)
        return answer


def _ok(body):
    return _Resp(200, body)


def _msg(mid, *, content="", attachments=(), embeds=(), **extra):
    return {
        "id": str(mid), "type": 0, "content": content,
        "timestamp": "2026-09-20T12:00:00.000000+00:00",
        "author": {"id": "7", "username": "artist"},
        "attachments": list(attachments), "embeds": list(embeds), **extra,
    }


def _client(routes, **kw):
    return DiscordClient("tok", session=_Session(routes), **kw)


def _ids(client, url, cursor=None):
    return [(m["id"], cur) for m, _meta, cur in client.iter_posts(url, cursor)]


# -- pure helpers --------------------------------------------------------------

def test_source_urls():
    assert parse_source_url("https://discord.com/channels/1/2") == ("1", "2")
    assert parse_source_url("https://discord.com/channels/1") == ("1", None)
    assert parse_source_url("https://discord.com/channels/1/2/threads/3") == ("1", "3")
    assert parse_source_url("https://discord.com/channels/@me/5") == (None, "5")
    assert parse_source_url("discord.com/channels/1/2/") == ("1", "2")


def test_a_message_link_is_not_a_source():
    with pytest.raises(DiscordAPIError):
        parse_source_url("https://discord.com/channels/1/2/3")
    with pytest.raises(DiscordAPIError):
        parse_source_url("https://example.com/channels/1/2")


def test_name_split_matches_gallery_dl():
    url = "https://cdn.discordapp.com/attachments/1/2/My%20Pic.final.PNG?ex=a&hm=b"
    assert nameext_from_url(url) == ("My Pic.final", "png")
    assert nameext_from_url("https://x/y/noext") == ("noext", "")
    assert nameext_from_url("https://x/y/a." + "b" * 17) == ("a." + "b" * 17, "")


def test_user_agent_is_gallery_dls_dated_firefox():
    """gallery-dl's own comment: "147 on 2026-01-13"."""
    assert "Firefox/147.0" in firefox_user_agent(date(2026, 1, 13))


def test_message_text_takes_rich_embeds_and_polls():
    m = _msg(1, content="hello", embeds=[
        {"type": "rich", "author": {"name": "A"}, "title": "T",
         "fields": [{"name": "f", "value": "v"}], "footer": {"text": "ft"}},
        {"type": "image", "title": "not text"},
    ], poll={"question": {"text": "Q?"}, "answers": [{"poll_media": {"text": "yes"}}]})
    assert message_text(m) == "hello\nA\nT\nf\nv\nft\nQ?\nyes"


def test_files_are_attachments_then_embeds_then_snapshots_numbered_across():
    m = _msg(9, attachments=[{"url": "https://cdn/a/1.png"}], embeds=[
        {"type": "video", "video": {"proxy_url": "https://media/v.mp4"},
         "thumbnail": {"proxy_url": "https://media/t.jpg"}},
        {"type": "image", "thumbnail": {"proxy_url": "https://media/i.webp"}},
        {"type": "rich", "image": {"proxy_url": "https://media/rich.png"}},
    ], message_snapshots=[
        {"message": {"type": 0, "attachments": [{"url": "https://cdn/a/fwd.gif"}],
                     "embeds": []}},
        {"message": {"type": 7, "attachments": [{"url": "https://cdn/a/join.png"}]}},
    ])
    items = DiscordClient.extract_media(m)
    assert [(i.num, i.filename, i.extension, i.kind) for i in items] == [
        (1, "1", "png", "attachment"),
        (2, "v", "mp4", "embed"),
        (3, "i", "webp", "embed"),
        (4, "fwd", "gif", "attachment"),
    ]
    assert {i.post_id for i in items} == {"9"}


def test_the_ledger_identity_survives_a_renumbering_edit():
    """Removing the first file renumbers the second; its identity must not move."""
    a = {"id": "100", "url": "https://cdn/a/1.png?ex=1"}
    b = {"id": "200", "url": "https://cdn/a/2.png?ex=1"}
    before = DiscordClient.extract_media(_msg(9, attachments=[a, b]))
    after = DiscordClient.extract_media(_msg(9, attachments=[b]))
    assert (before[1].num, after[0].num) == (2, 1)
    assert before[1].media_id == after[0].media_id == "200"


def test_an_embeds_identity_ignores_its_signature():
    def embed(sig):
        return {"type": "image", "image": {"proxy_url": f"https://media/p/x.png?ex={sig}"}}

    art = [{"url": "https://cdn/a/1.png"}]
    one = DiscordClient.extract_media(_msg(9, attachments=art, embeds=[embed("a")]))[1]
    two = DiscordClient.extract_media(_msg(9, attachments=art, embeds=[embed("b")]))[1]
    assert one.kind == two.kind == "embed"
    assert one.media_id == two.media_id and len(one.media_id) <= 33


def test_post_seams():
    with_file = _msg(5, attachments=[{"url": "https://cdn/a/1.png"}])
    assert DiscordClient.post_record_key(with_file) == ("message:5", "5")
    assert DiscordClient.post_record_key({}) is None


def test_a_text_only_message_is_not_a_post():
    """Chat lines would bury the drops."""
    assert DiscordClient.post_record_key(_msg(6, content="brb")) is None
    assert DiscordClient.post_meta(_msg(1))["date"].startswith("2026-09-20")


def test_a_message_without_an_attached_image_is_chat():
    """Operator 2026-09-28: only content with an image attached. A lone
    archive, a link preview or a Tenor GIF takes nothing and records nothing."""
    chat = [
        _msg(1, content="the stash", attachments=[
            {"url": "https://cdn/a/Links_Stash.rar", "content_type": "application/x-rar"},
        ]),
        _msg(2, content="look", embeds=[
            {"type": "image", "image": {"proxy_url": "https://media/p/x.png"}},
        ]),
        _msg(3, content="lol", embeds=[
            {"type": "gifv", "video": {"proxy_url": "https://media/p/t.mp4"}},
        ]),
        _msg(4, content="wip", attachments=[{"url": "https://cdn/a/wip.psd"}]),
    ]
    for m in chat:
        assert DiscordClient.extract_media(m) == []
        assert DiscordClient.post_record_key(m) is None


def test_an_attached_image_takes_every_file_numbered_as_gallery_dl_does():
    """The gate decides WHETHER a message is taken, never which of its files:
    the rar beside the image keeps its number, so on-disk names still match."""
    m = _msg(7, attachments=[
        {"url": "https://cdn/a/pack.rar"},
        {"url": "https://cdn/a/noext", "content_type": "image/png"},
    ])
    items = DiscordClient.extract_media(m)
    assert [(i.num, i.filename) for i in items] == [(1, "pack"), (2, "noext")]
    assert DiscordClient.post_record_key(m) == ("message:7", "7")
    video = _msg(8, attachments=[{"url": "https://cdn/a/clip.MP4?ex=1"}])
    assert len(DiscordClient.extract_media(video)) == 1


def test_a_forwarded_image_counts_as_attached():
    m = _msg(9, message_snapshots=[
        {"message": {"type": 0, "attachments": [{"url": "https://cdn/a/fwd.png"}]}},
    ])
    assert DiscordClient.post_record_key(m) == ("message:9", "9")


# -- the walk --------------------------------------------------------------------

def test_a_channel_pages_newest_first_and_skips_system_messages():
    page1 = [_msg(i) for i in range(300, 200, -1)]
    page1[3]["type"] = 7  # a member-join line: not content
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "S"}),
        "/guilds/1/channels": _ok([{"id": "2", "type": 0, "name": "art"}]),
        ("/channels/2/messages", None): _ok(page1),
        ("/channels/2/messages", "201"): _ok([_msg(150)]),
        ("/channels/2/threads/search", 0): _ok({"threads": []}),
    }
    got = _ids(_client(routes), "https://discord.com/channels/1/2")
    assert len(got) == 100  # 99 of page 1 + 1 of page 2
    assert "297" not in [mid for mid, _ in got]
    assert got[0] == ("300", "2:")
    assert got[-1] == ("150", "2:201")


def test_messages_carry_server_and_channel_metadata():
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "Studio", "owner_id": "9"}),
        "/guilds/1/channels": _ok([
            {"id": "4", "type": 4, "name": "Art"},
            {"id": "2", "type": 0, "name": "drops", "parent_id": "4"},
        ]),
        ("/channels/2/messages", None): _ok([_msg(10)]),
        ("/channels/2/threads/search", 0): _ok({"threads": []}),
    }
    [(msg, meta, _)] = list(_client(routes).iter_posts("https://discord.com/channels/1/2"))
    assert msg["_meta"] is meta
    assert meta["server"] == "Studio" and meta["server_id"] == "1"
    assert meta["channel"] == "drops" and meta["channel_id"] == "2"
    assert meta["parent"] == "Art"


def _one_channel(messages):
    return {
        "/guilds/1": _ok({"id": "1", "name": "Todding's Server"}),
        "/guilds/1/channels": _ok([
            {"id": "4", "type": 4, "name": "Todding"},
            {"id": "2", "type": 0, "name": "banana-land", "parent_id": "4"},
        ]),
        ("/channels/2/messages", None): _ok(messages),
        ("/channels/2/threads/search", 0): _ok({"threads": []}),
    }


def test_only_from_takes_the_creators_messages_and_no_one_elses():
    """Operator 2026-09-28: in the creator's server, another member posting a
    meme is not the creator's art. Matched by id, username or display name."""
    creator = {"id": "7", "username": "todding", "global_name": "Todding"}
    other = {"id": "8", "username": "jakeboii", "global_name": "Jake Boii"}
    msgs = [_msg(3, author=creator), _msg(2, author=other), _msg(1, author=creator)]
    url = "https://discord.com/channels/1/2"

    everyone = _client(_one_channel([dict(m) for m in msgs]))
    assert [mid for mid, _ in _ids(everyone, url)] == ["3", "2", "1"]
    for who in (["Todding"], ["TODDING "], ["7"]):
        client = _client(_one_channel([dict(m) for m in msgs]))
        client.only_from(who)
        assert [mid for mid, _ in _ids(client, url)] == ["3", "1"]
    client = _client(_one_channel([dict(m) for m in msgs]))
    client.only_from([])
    assert len(_ids(client, url)) == 3


def test_the_source_label_names_the_server_and_channel_the_walk_read():
    client = _client(_one_channel([]))
    url = "https://discord.com/channels/1/2"
    assert client.source_label(url) is None  # nothing walked yet
    list(client.iter_posts(url))
    assert client.source_label(url) == "Todding's Server · #banana-land"
    assert client.source_label("https://discord.com/channels/1") == "Todding's Server"
    client._channels["9"] = {"channel": "wip", "is_thread": True, "parent": "banana-land"}
    assert client.source_label("https://discord.com/channels/1/9") == (
        "Todding's Server · #banana-land › wip"
    )


def test_a_server_walks_text_then_threads_newest_created_first_and_skips_private():
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "S"}),
        "/guilds/1/channels": _ok([
            {"id": "2", "type": 0, "name": "text"},
            {"id": "3", "type": 2, "name": "voice"},
            {"id": "5", "type": 15, "name": "forum"},
            {"id": "6", "type": 0, "name": "private"},
        ]),
        ("/channels/2/messages", None): _ok([_msg(20)]),
        ("/channels/2/threads/search", 0): _ok({"threads": [
            {"id": "21", "type": 11, "name": "old", "parent_id": "2", "thread_metadata": {}},
            {"id": "22", "type": 11, "name": "new", "parent_id": "2", "thread_metadata": {}},
        ]}),
        ("/channels/22/messages", None): _ok([_msg(220)]),
        ("/channels/21/messages", None): _Resp(403, {}),  # a private thread
        ("/channels/5/threads/search", 0): _ok({"threads": [
            {"id": "51", "type": 11, "name": "post", "parent_id": "5", "thread_metadata": {}},
        ]}),
        ("/channels/51/messages", None): _ok([_msg(510)]),
        ("/channels/6/messages", None): _Resp(403, {}),
        ("/channels/6/threads/search", 0): _Resp(403, {}),
    }
    got = [mid for mid, _ in _ids(_client(routes), "https://discord.com/channels/1")]
    assert got == ["20", "220", "510"]


def test_a_resume_cursor_reenters_its_channel_at_its_page():
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "S"}),
        "/guilds/1/channels": _ok([
            {"id": "2", "type": 0, "name": "a"},
            {"id": "3", "type": 0, "name": "b"},
        ]),
        ("/channels/2/threads/search", 0): _ok({"threads": []}),
        ("/channels/3/threads/search", 0): _ok({"threads": []}),
        ("/channels/3/messages", "77"): _ok([_msg(70)]),
    }
    client = _client(routes)
    assert _ids(client, "https://discord.com/channels/1", "3:77") == [("70", "3:77")]
    fetched = [c for c in client._session.calls if c[0].endswith("/messages")]
    assert fetched == [("/channels/3/messages", {"limit": 100, "before": "77"})]


def test_skip_feed_ends_the_channel_not_the_walk():
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "S"}),
        "/guilds/1/channels": _ok([
            {"id": "2", "type": 0, "name": "a"},
            {"id": "3", "type": 0, "name": "b"},
        ]),
        ("/channels/2/messages", None): _ok([_msg(29), _msg(28)]),
        ("/channels/2/threads/search", 0): _ok({"threads": []}),
        ("/channels/3/messages", None): _ok([_msg(39)]),
        ("/channels/3/threads/search", 0): _ok({"threads": []}),
    }
    client = _client(routes)
    seen = []
    for msg, _meta, _cur in client.iter_posts("https://discord.com/channels/1"):
        seen.append(msg["id"])
        if msg["id"] == "29":
            client.skip_feed()
    assert seen == ["29", "39"]


def test_the_named_channel_refusing_the_token_is_an_auth_failure():
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "S"}),
        "/guilds/1/channels": _ok([{"id": "2", "type": 0, "name": "a"}]),
        ("/channels/2/messages", None): _Resp(403, {}),
    }
    with pytest.raises(DiscordAuthError):
        list(_client(routes).iter_posts("https://discord.com/channels/1/2"))


def test_401_is_an_invalid_token():
    with pytest.raises(DiscordAuthError):
        list(_client({"/guilds/1": _Resp(401, {})}).iter_posts(
            "https://discord.com/channels/1"))


def test_no_token_fails_before_any_request():
    client = DiscordClient(None, session=_Session({}))
    with pytest.raises(DiscordAuthError):
        list(client.iter_posts("https://discord.com/channels/1"))
    assert client._session.calls == []


def test_429_waits_and_retries(monkeypatch):
    waits = []
    monkeypatch.setattr(dc.time, "sleep", waits.append)
    routes = {"/users/@me": [
        _Resp(429, {}, {"Retry-After": "1.5"}), _ok({"username": "me"}),
    ], "/channels/2": _ok({"id": "2", "type": 0})}
    ok, msg = _client(routes).verify_auth("https://discord.com/channels/1/2")
    assert ok is True and "me" in msg
    assert waits == [1.5]


def test_verify_tells_a_bad_token_from_a_hidden_channel():
    bad = _client({"/users/@me": _Resp(401, {})})
    assert bad.verify_auth("https://discord.com/channels/1/2")[0] is False
    hidden = _client({"/users/@me": _ok({"username": "me"}),
                      "/channels/2": _Resp(403, {})})
    ok, msg = hidden.verify_auth("https://discord.com/channels/1/2")
    assert ok is False and "cannot see this channel" in msg
    assert _client({}).verify_auth("https://discord.com/channels/1/2/3")[0] is None


def test_the_request_profile_is_gallery_dls():
    client = _client({})
    h = client._session.headers
    assert h["Authorization"] == "tok"
    assert h["Referer"] == "https://discord.com/"
    assert h["Accept"] == "*/*"
    assert h["User-Agent"].startswith("Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:")


# -- the adapter ---------------------------------------------------------------

def test_the_adapter_authenticates_with_the_token_and_keys_by_identity(tmp_path):
    from backend.app.services.discord_ingester import DiscordIngester, _ledger_key

    ing = DiscordIngester(tmp_path, None, session_factory=None, auth_token="tok")
    assert ing.client._session.headers["Authorization"] == "tok"
    # A files-only drop is ordinary on Discord, not a broken parser.
    assert ing._body_canary is False
    [media] = DiscordClient.extract_media(
        _msg(9, attachments=[{"id": "300", "url": "https://cdn/a/1.png"}])
    )
    assert _ledger_key(media) == "9:300"


# -- the poster picker (#4491) -------------------------------------------------

def _person(pid, username, global_name=None, **extra):
    return {"id": pid, "username": username, "global_name": global_name,
            "messages": 1, "images": 0, **extra}


def test_the_creator_a_server_is_named_for_is_suggested():
    ranked = dc.rank_posters("Todding's Server", "99", [
        _person("8", "jakeboii", "Jake Boii", images=3),
        _person("7", "todding", "Todding", images=1),
    ])
    assert ranked[0]["id"] == "7" and ranked[0]["suggested"] is True
    assert ranked[0]["reasons"] == ["name matches the server"]
    assert ranked[1]["suggested"] is False


def test_owning_the_server_and_matching_its_name_outranks_either_alone():
    ranked = dc.rank_posters("The Official Todding Discord", "7", [
        _person("5", "toddingfan"),
        _person("7", "t0dd", "Todding"),
    ])
    assert [r["id"] for r in ranked if r["suggested"]] == ["7"]
    assert ranked[0]["reasons"] == ["owns the server", "name matches the server"]


def test_no_signal_suggests_no_one_and_bots_never():
    assert not any(r["suggested"] for r in dc.rank_posters("Art Club", None, [
        _person("1", "alice"), _person("2", "bob"),
    ]))
    ranked = dc.rank_posters("Todding's Server", None, [_person("3", "toddingbot", bot=True)])
    assert ranked[0]["suggested"] is False


def test_recent_posters_tallies_a_shallow_window_by_id():
    creator = {"id": "7", "username": "todding", "global_name": "Todding"}
    other = {"id": "8", "username": "jakeboii", "global_name": "Jake Boii"}
    art = [{"url": "https://cdn/a/1.png"}]
    page = [_msg(i, author=creator, attachments=art) for i in range(300, 240, -1)]
    page += [_msg(i, author=other) for i in range(240, 200, -1)]
    routes = {
        "/guilds/1": _ok({"id": "1", "name": "Todding's Server", "owner_id": "7"}),
        ("/channels/2/messages", None): _ok(page),
        ("/channels/2/messages", "201"): _ok([_msg(150, author=other)]),
    }
    client = _client(routes)
    got = client.recent_posters("1", "2", max_messages=150)
    assert got["server"] == "Todding's Server" and got["owner_id"] == "7"
    assert got["scanned"] == 101
    first = got["posters"][0]
    assert (first["id"], first["messages"], first["images"], first["suggested"]) == ("7", 60, 60, True)
    assert got["posters"][1]["messages"] == 41
    # It asked for no more than the window: 100, then the 50 left of 150.
    limits = [p.get("limit") for e, p in client._session.calls if e.endswith("/messages")]
    assert limits == [100, 50]