"""
Discord REST client, cache and settings (server-26#225 slice S2).

Every request goes through httpx.MockTransport: nothing here talks to Discord.
"""
import asyncio
import logging
from types import SimpleNamespace

import httpx
import pytest

from app.config import settings
from app.internal import discord_rest as rest
from tests.fake_firestore import FakeFirestore

TOKEN = "Nzk5.SECRET-TOKEN-VALUE"
G = "100000000000000001"
BOT = "900000000000000009"


class Clock:
    def __init__(self):
        self.t = 1000.0

    def __call__(self):
        return self.t


class Discord:
    """A scripted Discord: ``routes[(method, path)]`` is a Response or a
    callable(request) -> Response. Records every request."""

    def __init__(self):
        self.routes: dict = {}
        self.calls: list[httpx.Request] = []

    def __call__(self, request: httpx.Request) -> httpx.Response:
        self.calls.append(request)
        path = request.url.path.removeprefix("/api/v10")
        handler = self.routes.get((request.method, path))
        if handler is None:
            return httpx.Response(404, json={"code": 0, "message": "no route in test"})
        return handler(request) if callable(handler) else handler

    def paths(self):
        return [r.url.path.removeprefix("/api/v10") for r in self.calls]


@pytest.fixture
def discord(monkeypatch):
    d = Discord()
    rest.reset()
    monkeypatch.setattr(rest, "_transport", httpx.MockTransport(d))
    monkeypatch.setattr(rest, "_client_obj", None)
    clock = Clock()
    monkeypatch.setattr(rest, "time", SimpleNamespace(monotonic=clock, time=clock))
    slept: list = []

    async def _sleep(s):
        slept.append(s)
        clock.t += s
    monkeypatch.setattr(rest, "_sleep", _sleep)
    d.clock, d.slept = clock, slept
    yield d
    rest.reset()


@pytest.fixture
def fs(monkeypatch):
    return FakeFirestore.install(monkeypatch)


def _token_doc(fs, token_id="t1", *, guilds=None, valid=True, bot_user_id=BOT):
    doc = {"token_id": token_id, "name": "Radio 1", "token": TOKEN, "org_id": None,
           "guild_ids": [G] if guilds is None else guilds, "valid": valid, "bot_user_id": bot_user_id}
    fs.put("bot_tokens", token_id, doc)
    return dict(doc)


def _guild_routes(d, guild=G):
    d.routes[("GET", f"/guilds/{guild}")] = httpx.Response(200, json={"id": guild, "owner_id": "1", "roles": []})
    d.routes[("GET", f"/guilds/{guild}/members/{BOT}")] = httpx.Response(200, json={"roles": []})
    d.routes[("GET", f"/guilds/{guild}/channels")] = httpx.Response(200, json=[{"id": "5", "type": 2}])


# --- client hygiene ----------------------------------------------------------

async def test_requests_carry_user_agent_and_bot_auth(discord):
    discord.routes[("GET", "/users/@me")] = httpx.Response(200, json={"id": BOT})
    r = await rest.request(TOKEN, "GET", "/users/@me")
    assert r.ok and r.data == {"id": BOT}
    (req,) = discord.calls
    assert req.headers["User-Agent"] == "DiscordBot (https://drb.cusano.net, 1)"
    assert req.headers["Authorization"] == f"Bot {TOKEN}"


async def test_token_never_logged_or_in_errors(discord, fs, caplog):
    caplog.set_level(logging.DEBUG)
    discord.routes[("GET", "/users/@me")] = httpx.Response(500, json={"message": "boom"})
    info = await rest.inspect_token(TOKEN)
    assert info["valid"] is None and info["check_error"]

    def _boom(request):
        raise httpx.ConnectError("connection refused", request=request)
    discord.routes[("GET", f"/guilds/{G}")] = _boom
    snap = await rest.guild_snapshot(_token_doc(fs), G)
    assert snap.status == rest.UNKNOWN
    assert TOKEN not in (info["check_error"] or "") and TOKEN not in (snap.error or "")
    assert TOKEN not in caplog.text and "SECRET" not in caplog.text


# --- /users/@me/guilds pagination --------------------------------------------

def _guild_pages(d, total):
    ids = [str(10**17 + i) for i in range(total)]

    def _page(request):
        after = request.url.params.get("after")
        limit = int(request.url.params["limit"])
        rest_ids = [i for i in ids if after is None or int(i) > int(after)]
        return httpx.Response(200, json=[{"id": i} for i in rest_ids[:limit]])
    d.routes[("GET", "/users/@me/guilds")] = _page
    return ids


async def test_guild_list_follows_after_pages(discord):
    ids = _guild_pages(discord, 450)
    r = await rest.list_guild_ids(TOKEN)
    assert r.ok and r.data == ids
    pages = [c for c in discord.calls if c.url.path.endswith("/users/@me/guilds")]
    assert len(pages) == 3
    assert pages[0].url.params.get("after") is None
    assert pages[1].url.params["after"] == ids[199] and pages[2].url.params["after"] == ids[399]


async def test_guild_list_exact_multiple_of_page_size_stops_on_empty_page(discord):
    ids = _guild_pages(discord, 400)
    r = await rest.list_guild_ids(TOKEN)
    assert r.data == ids and len(discord.calls) == 3


async def test_failed_page_is_unknown_not_a_truncated_list(discord):
    ids = [str(10**17 + i) for i in range(200)]

    def _page(request):
        if request.url.params.get("after"):
            return httpx.Response(502)
        return httpx.Response(200, json=[{"id": i} for i in ids])
    discord.routes[("GET", "/users/@me/guilds")] = _page
    r = await rest.list_guild_ids(TOKEN)
    assert not r.ok and r.unknown and r.data is None


async def test_inspect_token_reports_all_servers(discord):
    ids = _guild_pages(discord, 250)
    discord.routes[("GET", "/users/@me")] = httpx.Response(200, json={"id": BOT, "username": "DRB Radio 1"})
    discord.routes[("GET", "/oauth2/applications/@me")] = httpx.Response(200, json={"id": "777"})
    info = await rest.inspect_token(TOKEN)
    assert info["valid"] is True and info["guild_ids"] == ids
    assert info["bot_user_id"] == BOT and info["bot_name"] == "DRB Radio 1" and info["application_id"] == "777"


async def test_inspect_token_401_raises_and_5xx_is_unknown(discord):
    discord.routes[("GET", "/users/@me")] = httpx.Response(401, json={"code": 0})
    with pytest.raises(rest.TokenRejected):
        await rest.inspect_token(TOKEN)
    discord.routes[("GET", "/users/@me")] = httpx.Response(503)
    info = await rest.inspect_token(TOKEN)
    assert info["valid"] is None and info["guild_ids"] is None and "503" in info["check_error"]


async def test_refresh_membership_writes_full_list_and_drops_cache(discord, fs):
    doc = _token_doc(fs, guilds=["1"])
    ids = _guild_pages(discord, 201)
    discord.routes[("GET", "/users/@me")] = httpx.Response(200, json={"id": BOT, "username": "r"})
    rest.access_cache._entries[("t1", G, "snapshot")] = (10**9, "cached")
    assert await rest.refresh_membership(["t1"]) == 1
    assert fs.get("bot_tokens", "t1")["guild_ids"] == ids
    assert rest.access_cache.peek(("t1", G, "snapshot")) is None
    assert doc["token"] == fs.get("bot_tokens", "t1")["token"]


# --- rate limits ---------------------------------------------------------------

async def test_short_429_is_retried_once(discord):
    seq = [httpx.Response(429, json={"retry_after": 1.5, "global": False}),
           httpx.Response(200, json={"id": BOT})]
    discord.routes[("GET", "/users/@me")] = lambda req: seq.pop(0)
    r = await rest.request(TOKEN, "GET", "/users/@me")
    assert r.ok and discord.slept == [1.5] and len(discord.calls) == 2


async def test_second_429_or_long_retry_after_is_unknown(discord):
    discord.routes[("GET", "/users/@me")] = httpx.Response(429, json={"retry_after": 1.0})
    r = await rest.request(TOKEN, "GET", "/users/@me")
    assert r.status == 429 and r.unknown and len(discord.calls) == 2

    discord.calls.clear()
    discord.slept.clear()
    discord.routes[("GET", "/users/@me")] = httpx.Response(429, json={"retry_after": 30})
    r = await rest.request(TOKEN, "GET", "/users/@me")
    assert r.unknown and len(discord.calls) == 1 and discord.slept == []


async def test_global_429_pauses_only_that_token(discord):
    discord.routes[("GET", "/users/@me")] = httpx.Response(
        429, json={"retry_after": 60, "global": True}, headers={"X-RateLimit-Global": "true"})
    assert (await rest.request(TOKEN, "GET", "/users/@me")).unknown
    discord.routes[("GET", "/users/@me")] = httpx.Response(200, json={"id": BOT})
    r = await rest.request(TOKEN, "GET", "/users/@me")
    assert r.status is None and "rate limited" in r.error and len(discord.calls) == 1
    assert (await rest.request("other-token", "GET", "/users/@me")).ok
    discord.clock.t += 61
    assert (await rest.request(TOKEN, "GET", "/users/@me")).ok


async def test_exhausted_bucket_is_per_token_and_guild(discord):
    other = "200000000000000002"
    hdrs = {"X-RateLimit-Bucket": "abc", "X-RateLimit-Remaining": "0", "X-RateLimit-Reset-After": "60"}
    discord.routes[("GET", f"/guilds/{G}/channels")] = httpx.Response(200, json=[], headers=hdrs)
    discord.routes[("GET", f"/guilds/{other}/channels")] = httpx.Response(200, json=[], headers=hdrs)
    assert (await rest.request(TOKEN, "GET", f"/guilds/{G}/channels")).ok
    blocked = await rest.request(TOKEN, "GET", f"/guilds/{G}/channels")
    assert blocked.status is None and len(discord.calls) == 1        # not sent
    assert (await rest.request(TOKEN, "GET", f"/guilds/{other}/channels")).ok   # other guild
    assert (await rest.request("tok-2", "GET", f"/guilds/{G}/channels")).ok     # other token
    discord.clock.t += 61
    assert (await rest.request(TOKEN, "GET", f"/guilds/{G}/channels")).ok


async def test_short_bucket_wait_is_slept_not_refused(discord):
    hdrs = {"X-RateLimit-Bucket": "abc", "X-RateLimit-Remaining": "0", "X-RateLimit-Reset-After": "2"}
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(200, json={}, headers=hdrs)
    await rest.request(TOKEN, "GET", f"/guilds/{G}")
    assert (await rest.request(TOKEN, "GET", f"/guilds/{G}")).ok and discord.slept == [2.0]


# --- invalid-request breaker -----------------------------------------------------

async def test_breaker_stops_all_requests_and_drains(discord, monkeypatch):
    monkeypatch.setattr(settings, "discord_rest_invalid_budget", 3)
    discord.routes[("GET", "/users/@me")] = httpx.Response(401)
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(
        429, json={"retry_after": 99}, headers={"X-RateLimit-Scope": "shared"})
    for _ in range(4):                       # shared 429s never count
        await rest.request("t-x", "GET", f"/guilds/{G}")
    for _ in range(3):
        assert (await rest.request(TOKEN, "GET", "/users/@me")).status == 401
    sent = len(discord.calls)
    r = await rest.request("a-fresh-token", "GET", "/users/@me")
    assert r.status is None and "paused" in r.error and len(discord.calls) == sent
    discord.clock.t += settings.discord_rest_invalid_window_seconds + 1
    assert (await rest.request("a-fresh-token", "GET", "/users/@me")).status == 401


async def test_403_and_unshared_429_count_toward_breaker(discord, monkeypatch):
    monkeypatch.setattr(settings, "discord_rest_invalid_budget", 2)
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(403, json={"code": 50013})
    discord.routes[("GET", "/users/@me")] = httpx.Response(429, json={"retry_after": 99})
    await rest.request(TOKEN, "GET", f"/guilds/{G}")
    await rest.request("t2", "GET", "/users/@me")
    assert (await rest.request("t3", "GET", "/users/@me")).error.startswith("Discord REST paused")


# --- per-(token, guild) snapshot + cache ---------------------------------------------

async def test_snapshot_reads_three_endpoints_and_caches(discord, fs, monkeypatch):
    _guild_routes(discord)
    tok = _token_doc(fs)
    snap = await rest.guild_snapshot(tok, G)
    assert snap.status == rest.OK and snap.channels == [{"id": "5", "type": 2}] and snap.member == {"roles": []}
    assert sorted(discord.paths()) == sorted([f"/guilds/{G}", f"/guilds/{G}/members/{BOT}", f"/guilds/{G}/channels"])
    assert (await rest.guild_snapshot(tok, G)) is snap and len(discord.calls) == 3
    discord.clock.t += settings.discord_access_ttl_seconds + 1
    await rest.guild_snapshot(tok, G)
    assert len(discord.calls) == 6
    assert rest.invalidate(guild_id=G) == 1
    await rest.guild_snapshot(tok, G)
    assert len(discord.calls) == 9
    await rest.guild_snapshot(tok, G, refresh=True)
    assert len(discord.calls) == 12


async def test_snapshot_learns_and_stores_bot_user_id(discord, fs):
    _guild_routes(discord)
    discord.routes[("GET", "/users/@me")] = httpx.Response(200, json={"id": BOT})
    snap = await rest.guild_snapshot(_token_doc(fs, bot_user_id=None), G)
    assert snap.status == rest.OK and snap.bot_user_id == BOT
    assert fs.get("bot_tokens", "t1")["bot_user_id"] == BOT


async def test_snapshot_401_marks_token_invalid_and_is_not_retried(discord, fs):
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(401, json={"code": 0})
    tok = _token_doc(fs)
    snap = await rest.guild_snapshot(tok, G)
    assert snap.status == rest.INVALID and fs.get("bot_tokens", "t1")["valid"] is False
    assert len(discord.calls) == 1
    discord.clock.t += settings.discord_access_negative_ttl_seconds - 1
    assert (await rest.guild_snapshot(tok, G)).status == rest.INVALID and len(discord.calls) == 1
    # an invalid token doc is never sent to Discord at all
    rest.invalidate()
    assert (await rest.guild_snapshot({**tok, "valid": False}, G)).status == rest.INVALID
    assert len(discord.calls) == 1


@pytest.mark.parametrize("code,path", [(10004, f"/guilds/{G}"), (10007, f"/guilds/{G}/members/{BOT}")])
async def test_unknown_guild_or_member_drops_membership(discord, fs, code, path):
    _guild_routes(discord)
    discord.routes[("GET", path)] = httpx.Response(404, json={"code": code})
    snap = await rest.guild_snapshot(_token_doc(fs, guilds=[G, "7"]), G)
    assert snap.status == rest.NOT_MEMBER
    assert fs.get("bot_tokens", "t1")["guild_ids"] == ["7"]


async def test_missing_access_confirms_with_one_membership_read(discord, fs):
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(403, json={"code": 50001})
    discord.routes[("GET", "/users/@me/guilds")] = httpx.Response(200, json=[{"id": "7"}])
    snap = await rest.guild_snapshot(_token_doc(fs, guilds=[G, "7"]), G)
    assert snap.status == rest.NOT_MEMBER
    assert fs.get("bot_tokens", "t1")["guild_ids"] == ["7"]
    assert discord.paths().count("/users/@me/guilds") == 1


async def test_missing_access_alone_never_changes_membership(discord, fs):
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(403, json={"code": 50001})
    discord.routes[("GET", "/users/@me/guilds")] = httpx.Response(200, json=[{"id": G}])
    snap = await rest.guild_snapshot(_token_doc(fs, guilds=[G]), G)
    assert snap.status == rest.UNKNOWN and fs.get("bot_tokens", "t1")["guild_ids"] == [G]

    rest.invalidate()
    discord.routes[("GET", "/users/@me/guilds")] = httpx.Response(503)   # confirmation unknown too
    snap = await rest.guild_snapshot(_token_doc(fs, guilds=[G]), G)
    assert snap.status == rest.UNKNOWN and fs.get("bot_tokens", "t1")["guild_ids"] == [G]


async def test_non_discord_failure_is_unknown_not_raised(discord, fs, monkeypatch):
    # A Firestore error while classifying (here: dropping membership) must
    # fail open — S3's start_stream relies on the snapshot never raising.
    _guild_routes(discord)
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(404, json={"code": 10004})
    tok = _token_doc(fs)

    async def _boom(*a, **kw):
        raise RuntimeError("firestore down")
    monkeypatch.setattr(rest, "_drop_membership", _boom)
    snap = await rest.guild_snapshot(tok, G)
    assert snap.status == rest.UNKNOWN and "RuntimeError" in snap.error
    assert TOKEN not in snap.error


async def test_unknown_result_is_cached_briefly(discord, fs):
    discord.routes[("GET", f"/guilds/{G}")] = httpx.Response(503)
    tok = _token_doc(fs)
    assert (await rest.guild_snapshot(tok, G)).status == rest.UNKNOWN
    await rest.guild_snapshot(tok, G)
    assert len(discord.calls) == 1
    discord.clock.t += settings.discord_access_unknown_ttl_seconds + 1
    await rest.guild_snapshot(tok, G)
    assert len(discord.calls) == 2


async def test_concurrent_loads_share_one_flight(discord, fs, monkeypatch):
    gate = asyncio.Event()
    _guild_routes(discord)
    tok = _token_doc(fs)
    real = rest._load_snapshot
    loads = []

    async def _slow(token_doc, guild_id):
        loads.append(guild_id)
        await gate.wait()
        return await real(token_doc, guild_id)
    monkeypatch.setattr(rest, "_load_snapshot", _slow)
    tasks = [asyncio.create_task(rest.guild_snapshot(tok, G)) for _ in range(5)]
    await asyncio.sleep(0)
    gate.set()
    snaps = await asyncio.gather(*tasks)
    assert loads == [G] and len(discord.calls) == 3
    assert all(s is snaps[0] for s in snaps)


async def test_invalidate_during_a_load_is_not_undone(discord, fs, monkeypatch):
    gate = asyncio.Event()
    _guild_routes(discord)
    tok = _token_doc(fs)
    real = rest._load_snapshot

    async def _slow(token_doc, guild_id):
        await gate.wait()
        return await real(token_doc, guild_id)
    monkeypatch.setattr(rest, "_load_snapshot", _slow)
    task = asyncio.create_task(rest.guild_snapshot(tok, G))
    await asyncio.sleep(0)
    rest.invalidate(token_id="t1")
    gate.set()
    assert (await task).status == rest.OK
    assert rest.access_cache.peek(("t1", G, "snapshot")) is None


def test_invalidate_scopes():
    c = rest.AccessCache()
    for k in [("t1", "g1", "s"), ("t1", "g2", "s"), ("t2", "g1", "s")]:
        c._entries[k] = (10**12, k)
    assert c.invalidate(token_id="t1", guild_id="g1") == 1
    assert c.invalidate(guild_id="g1") == 1
    assert c.invalidate(token_id="t1") == 1
    assert c.invalidate() == 0


def test_ttl_for_outcomes():
    assert rest.ttl_for(rest.OK) == 120
    assert rest.ttl_for(rest.DENIED) == rest.ttl_for(rest.NOT_MEMBER) == rest.ttl_for(rest.INVALID) == 600
    assert rest.ttl_for(rest.UNKNOWN) == 30


def test_225_settings_defaults():
    assert settings.discord_perm_check_enabled is True
    assert settings.discord_access_ttl_seconds == 120
    assert settings.discord_rest_invalid_budget == 4000
    assert settings.discord_oauth_redirect_uri == ""
    assert settings.discord_pool_bot_guild_cap == 90


# --- leave_guild on the shared client ---------------------------------------------

async def test_leave_guild_drops_cache_and_reports_unknown(discord, fs):
    _token_doc(fs, "t1", guilds=[G])
    fs.put("bot_tokens", "t2", {"token_id": "t2", "name": "R2", "token": "tok-two", "org_id": None,
                                "guild_ids": None, "valid": True})
    rest.access_cache._entries[("t1", G, "snapshot")] = (10**12, "x")

    def _delete(request):
        if request.headers["Authorization"] == "Bot tok-two":
            raise httpx.ReadTimeout("timed out", request=request)
        return httpx.Response(204)
    discord.routes[("DELETE", f"/users/@me/guilds/{G}")] = _delete
    rows = {r["token_id"]: r for r in await rest.leave_guild(G, None)}
    assert rows["t1"]["left"] is True and rows["t2"]["left"] is False
    assert "ReadTimeout" in rows["t2"]["error"] and "tok-two" not in rows["t2"]["error"]
    assert fs.get("bot_tokens", "t1")["guild_ids"] == []
    assert rest.access_cache.peek(("t1", G, "snapshot")) is None