"""
Per-bot OAuth "Connect a Discord server" (server-26#225 S6).

Discord is an httpx.MockTransport; Firestore is tests/fake_firestore.py.
"""
import asyncio
import base64
import json
import logging
from datetime import datetime, timedelta, timezone
from urllib.parse import parse_qs, urlparse

import httpx
import pytest
from fastapi import HTTPException

from app.config import settings
from app.internal import discord_guild_link as glink
from app.internal import discord_oauth as oauth
from app.internal import discord_rest
from app.internal import discord_streams as streams
from app.routers import discord_oauth as router
from app.routers import tokens as tokens_router
from tests.fake_firestore import FakeFirestore

REDIRECT = "https://drb.example.test/discord/connected"
SECRET_A = "s3cr3t-app-a-0123456789abcdefXYZ"
SECRET_POOL = "s3cr3t-pool-0123456789abcdefXYZ"
GUILD = "100000000000000001"
OWNER_A = {"uid": "owner-a", "email": "a@example.test", "org_id": "org-a", "org_role": "owner"}
OWNER_B = {"uid": "owner-b", "email": "b@example.test", "org_id": "org-b", "org_role": "owner"}
ADMIN = {"uid": "admin-1", "email": "admin@example.test", "role": "admin", "org_id": "founding"}


class Discord:
    """Scripted OAuth endpoints. ``token_reply`` answers POST /oauth2/token."""

    def __init__(self):
        self.calls: list[httpx.Request] = []
        self.token_reply = httpx.Response(200, json={
            "access_token": "user-access-token", "token_type": "Bearer", "scope": "bot",
            "guild": {"id": GUILD, "name": "Station 7"}})

    def __call__(self, request: httpx.Request) -> httpx.Response:
        self.calls.append(request)
        path = request.url.path.removeprefix("/api/v10")
        if path == "/oauth2/token":
            r = self.token_reply
            return r(request) if callable(r) else r
        if path == "/oauth2/token/revoke":
            return httpx.Response(200, json={})
        return httpx.Response(404, json={"code": 0})

    def form(self, i: int) -> dict:
        return {k: v[0] for k, v in parse_qs(self.calls[i].content.decode()).items()}

    def basic(self, i: int) -> str:
        raw = self.calls[i].headers["authorization"].split(" ", 1)[1]
        return base64.b64decode(raw).decode()

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


@pytest.fixture
def discord(monkeypatch):
    d = Discord()
    discord_rest.reset()
    monkeypatch.setattr(discord_rest, "_transport", httpx.MockTransport(d))
    return d


@pytest.fixture
def fs(monkeypatch, discord):
    fake = FakeFirestore.install(monkeypatch)
    monkeypatch.setattr(settings, "discord_oauth_redirect_uri", REDIRECT)
    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "")
    monkeypatch.setattr(settings, "discord_pool_bot_guild_cap", 90)
    for org in ("org-a", "org-b", "founding"):
        fake.put("organizations", org, {"org_id": org, "name": org})
    fake.put("org_members", "owner-a", {"uid": "owner-a", "org_id": "org-a", "org_role": "owner"})
    fake.put("org_members", "owner-b", {"uid": "owner-b", "org_id": "org-b", "org_role": "owner"})
    # org-a's own bot, two pool bots (one full), org-b's own bot.
    fake.put("bot_tokens", "own-a", {"token_id": "own-a", "name": "Riverton Radio", "token": "tok-own-a",
                                     "org_id": "org-a", "application_id": "111", "guild_ids": []})
    fake.put("bot_tokens", "pool-1", {"token_id": "pool-1", "name": "DRB Radio 1", "token": "tok-pool-1",
                                      "org_id": None, "application_id": "201", "guild_ids": ["9"] * 40})
    fake.put("bot_tokens", "pool-2", {"token_id": "pool-2", "name": "DRB Radio 2", "token": "tok-pool-2",
                                      "org_id": None, "application_id": "202", "guild_ids": ["9"] * 90})
    fake.put("bot_tokens", "own-b", {"token_id": "own-b", "name": "Harbor Radio", "token": "tok-own-b",
                                     "org_id": "org-b", "application_id": "333", "guild_ids": []})
    for tid, secret in (("own-a", SECRET_A), ("pool-1", SECRET_POOL), ("pool-2", SECRET_POOL),
                        ("own-b", SECRET_A)):
        fake.put(oauth.OAUTH, tid, {"token_id": tid, "client_secret": secret, "enabled": True})
    for lim in (router.start_limiter, router.complete_limiter):
        lim._log.clear()
    stopped: list = []

    async def _stop_guild(guild_id, reason, channel_id=None, stream_id=None):
        stopped.append((guild_id, reason))
        return [f"{guild_id}:1"]
    monkeypatch.setattr(streams, "stop_guild", _stop_guild)
    fake.stopped = stopped

    async def _voice_check(**kw):
        return {"enabled": True, "channels": [
            {"channel_id": "1", "name": "dispatch", "state": "ready", "bots": [{"fix": None}]},
            {"channel_id": "2", "name": "fireground", "state": "needs_fix",
             "bots": [{"fix": "In Discord: Edit Channel → Permissions → add Riverton Radio → allow Speak, "
                              "then try again."}]},
            {"channel_id": "3", "name": "ops", "state": "unknown", "bots": []},
        ]}
    monkeypatch.setattr(router, "voice_check", _voice_check)
    return fake


def _audit(fake, action):
    return [e for e in (fake.data.get("audit_log") or {}).values() if e["action"] == action]


def _status(resp):
    return getattr(resp, "status_code", 200)


def _body(resp):
    return json.loads(resp.body) if hasattr(resp, "body") else resp


async def _start(decoded=OWNER_A, **kw):
    out = await router.oauth_start(router.StartBody(**kw), org_id=None, decoded=decoded)
    q = parse_qs(urlparse(out["authorize_url"]).query)
    return out, {k: v[0] for k, v in q.items()}


async def _complete(state, decoded=OWNER_A, code="the-code"):
    return await router.oauth_complete(router.CompleteBody(code=code, state=state), decoded=decoded)


# ---------------------------------------------------------------------------
# Off switch
# ---------------------------------------------------------------------------

async def test_every_route_is_404_while_redirect_uri_is_empty(fs, monkeypatch):
    monkeypatch.setattr(settings, "discord_oauth_redirect_uri", "")
    for call in (router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A),
                 router.oauth_start(router.StartBody(), org_id=None, decoded=OWNER_A),
                 _complete("anything")):
        with pytest.raises(HTTPException) as e:
            await call
        assert e.value.status_code == 404


async def test_404_when_no_usable_bot_has_a_secret(fs):
    for tid in ("own-a", "pool-1", "pool-2"):
        fs.data[oauth.OAUTH].pop(tid)
    with pytest.raises(HTTPException) as e:
        await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
    assert e.value.status_code == 404


# ---------------------------------------------------------------------------
# Which bot
# ---------------------------------------------------------------------------

async def test_org_bot_first_then_least_full_pool_bot_under_cap(fs):
    out = await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
    assert out["bot"] == {"name": "Riverton Radio", "reason": "org_bot", "own_bot": True}
    fs.data[oauth.OAUTH].pop("own-a")
    out = await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
    assert out["bot"]["name"] == "DRB Radio 1" and out["bot"]["reason"] == "pool"
    # pool-2 is at the cap (90): never offered, even when it's the only one left
    fs.data[oauth.OAUTH].pop("pool-1")
    with pytest.raises(HTTPException) as e:
        await router.oauth_options(node_id=None, token_id=None, org_id=None, decoded=OWNER_A)
    assert e.value.status_code == 404


async def test_preferred_bot_for_the_nodes_system_wins(fs):
    fs.put("nodes", "n1", {"node_id": "n1", "org_id": "org-a", "assigned_system_id": "s1"})
    fs.put("systems", "s1", {"system_id": "s1", "org_id": "org-a", "preferred_token_id": "pool-1"})
    out = await router.oauth_options(node_id="n1", token_id=None, org_id=None, decoded=OWNER_A)
    assert out["bot"]["name"] == "DRB Radio 1" and out["bot"]["reason"] == "preferred"
    # another org's node is a 404
    fs.put("nodes", "nb", {"node_id": "nb", "org_id": "org-b", "assigned_system_id": "s1"})
    with pytest.raises(HTTPException) as e:
        await router.oauth_options(node_id="nb", token_id=None, org_id=None, decoded=OWNER_A)
    assert e.value.status_code == 404


async def test_explicit_bot_must_be_usable_by_the_org(fs):
    out = await router.oauth_options(node_id=None, token_id="pool-1", org_id=None, decoded=OWNER_A)
    assert out["bot"]["reason"] == "chosen"
    for tid in ("own-b", "pool-2", "nope"):     # other org's bot, full pool bot, unknown
        with pytest.raises(HTTPException) as e:
            await router.oauth_options(node_id=None, token_id=tid, org_id=None, decoded=OWNER_A)
        assert e.value.status_code == 404


# ---------------------------------------------------------------------------
# Start
# ---------------------------------------------------------------------------

async def test_start_returns_the_bot_invite_and_stores_only_a_hash_of_the_state(fs):
    out, q = await _start(return_to="/nodes/n1")
    assert urlparse(out["authorize_url"]).netloc == "discord.com"
    assert q["client_id"] == "111" and q["scope"] == "bot" and q["response_type"] == "code"
    assert q["permissions"] == str((1 << 10) | (1 << 20) | (1 << 21))
    assert q["redirect_uri"] == REDIRECT
    nonce = q["state"]
    assert len(nonce) >= 40
    assert fs.get(oauth.STATES, nonce) is None                       # the nonce itself is never stored
    st = fs.get(oauth.STATES, oauth.state_key(nonce))
    assert st["org_id"] == "org-a" and st["uid"] == "owner-a" and st["token_id"] == "own-a"
    assert st["return_to"] == "/nodes/n1" and st["used_at"] is None
    assert len(_audit(fs, "discord.oauth.start")) == 1


@pytest.mark.parametrize("bad", ["https://evil.test/x", "//evil.test/x", "/\\evil.test", "nodes", "/a\nb"])
async def test_return_to_is_a_same_site_path_only(fs, bad):
    _, q = await _start(return_to=bad)
    assert fs.get(oauth.STATES, oauth.state_key(q["state"]))["return_to"] is None


async def test_start_rate_limited_per_user(fs):
    for _ in range(5):
        await _start()
    with pytest.raises(HTTPException) as e:
        await _start()
    assert e.value.status_code == 429


# ---------------------------------------------------------------------------
# Complete
# ---------------------------------------------------------------------------

async def test_complete_links_the_server_from_the_exchange_and_reports_channels(fs, discord):
    _, q = await _start(return_to="/settings/discord")
    out = await _complete(q["state"])
    assert out["ok"] and out["guild_id"] == GUILD and out["guild_name"] == "Station 7"
    assert out["org_id"] == "org-a" and out["bot_name"] == "Riverton Radio"
    assert (out["channels_ready"], out["channels_blocked"]) == (1, 1)
    assert out["channels"][1]["fix"].startswith("In Discord")
    assert out["return_to"] == "/settings/discord"

    g = fs.get(streams.GUILDS, GUILD)
    assert g["org_id"] == "org-a" and g["linked_via"] == "oauth" and g["linked_by"] == "owner-a"
    assert g["linked_with_token_id"] == "own-a"
    assert fs.get("bot_tokens", "own-a")["guild_ids"] == [GUILD]
    # the exchange used that bot application's own credentials and the exact redirect URI
    assert discord.paths() == ["/oauth2/token", "/oauth2/token/revoke"]
    assert discord.basic(0) == f"111:{SECRET_A}"
    assert discord.form(0) == {"grant_type": "authorization_code", "code": "the-code", "redirect_uri": REDIRECT}
    assert discord.form(1)["token"] == "user-access-token"           # revoked straight away
    assert fs.get(oauth.STATES, oauth.state_key(q["state"]))["used_at"]
    [entry] = _audit(fs, "discord.guild.link")
    assert entry["details"]["via"] == "oauth" and entry["details"]["guild_id"] == GUILD


async def test_state_is_single_use(fs):
    _, q = await _start()
    assert _status(await _complete(q["state"])) == 200
    again = await _complete(q["state"])
    assert _status(again) == 409 and _body(again)["code"] == "state_used"


async def test_another_users_state_is_refused_and_burned(fs):
    _, q = await _start(OWNER_A)
    resp = await _complete(q["state"], decoded=OWNER_B)              # login-CSRF shape
    assert _status(resp) == 403 and _body(resp)["code"] == "wrong_user"
    assert fs.get(streams.GUILDS, GUILD) is None
    assert _status(await _complete(q["state"], decoded=OWNER_A)) == 409   # consumed


async def test_expired_and_unknown_states(fs):
    _, q = await _start()
    key = oauth.state_key(q["state"])
    past = (datetime.now(timezone.utc) - timedelta(seconds=1)).isoformat()
    fs.put(oauth.STATES, key, {**fs.get(oauth.STATES, key), "expires_at": past})
    resp = await _complete(q["state"])
    assert _status(resp) == 410 and fs.get(streams.GUILDS, GUILD) is None
    assert _status(await _complete("not-a-state")) == 400


async def test_caller_who_is_no_longer_owner_is_refused(fs, discord):
    _, q = await _start()
    fs.put("org_members", "owner-a", {"uid": "owner-a", "org_id": "org-a", "org_role": "member"})
    resp = await _complete(q["state"])                               # token claims still say owner
    assert _status(resp) == 403 and _body(resp)["code"] == "not_owner"
    assert discord.calls == [] and fs.get(streams.GUILDS, GUILD) is None


async def test_exchange_without_a_guild_links_nothing(fs, discord):
    discord.token_reply = httpx.Response(200, json={"access_token": "x", "scope": "bot"})
    _, q = await _start()
    resp = await _complete(q["state"])
    assert _status(resp) == 400 and _body(resp)["code"] == "no_guild"
    assert fs.get(streams.GUILDS, GUILD) is None


async def test_rejected_code_and_unreachable_discord(fs, discord):
    discord.token_reply = httpx.Response(400, json={"error": "invalid_grant"})
    _, q = await _start()
    resp = await _complete(q["state"])
    assert _status(resp) == 400 and _body(resp)["code"] == "code_rejected"

    def _down(request):
        raise httpx.ConnectError("down", request=request)
    discord.token_reply = _down
    _, q = await _start()
    resp = await _complete(q["state"])
    assert _status(resp) == 502 and fs.get(streams.GUILDS, GUILD) is None


async def test_server_linked_to_another_org_is_never_moved(fs):
    fs.put(streams.GUILDS, GUILD, {"guild_id": GUILD, "org_id": "org-b", "guild_name": "Harbor"})
    _, q = await _start()
    resp = await _complete(q["state"])
    assert _status(resp) == 409 and _body(resp)["code"] == "linked_elsewhere"
    assert fs.get(streams.GUILDS, GUILD)["org_id"] == "org-b"
    assert fs.stopped == []
    assert [e["details"]["via"] for e in _audit(fs, "discord.guild.link_refused")] == ["oauth"]


async def test_linking_away_from_the_fallback_org_stops_its_streams(fs, monkeypatch):
    monkeypatch.setattr(settings, "discord_unlinked_guild_org_id", "founding")
    _, q = await _start()
    out = await _complete(q["state"])
    assert out["streams_stopped"] == 1 and fs.stopped == [(GUILD, glink.RELINK_STOP_REASON)]


async def test_permission_check_timeout_leaves_the_link_standing(fs, monkeypatch):
    monkeypatch.setattr(settings, "discord_oauth_link_check_seconds", 0.05)

    async def _slow(**kw):
        await asyncio.sleep(1)
    monkeypatch.setattr(router, "voice_check", _slow)
    _, q = await _start()
    out = await _complete(q["state"])
    assert out["ok"] and out["checked"] is False and out["channels_ready"] is None
    assert fs.get(streams.GUILDS, GUILD)["org_id"] == "org-a"


async def test_bot_handed_to_another_org_after_start_is_refused(fs):
    _, q = await _start()
    fs.put("bot_tokens", "own-a", {**fs.get("bot_tokens", "own-a"), "org_id": "org-b"})
    resp = await _complete(q["state"])
    assert _status(resp) == 409 and _body(resp)["code"] == "bot_unavailable"


# ---------------------------------------------------------------------------
# Secrets: PUT /tokens/{id}/oauth, never returned, never logged
# ---------------------------------------------------------------------------

async def test_set_secret_checks_it_with_discord_and_never_echoes_it(fs, discord, caplog):
    caplog.set_level(logging.DEBUG)
    fs.data[oauth.OAUTH].clear()
    discord.token_reply = httpx.Response(401, json={"error": "invalid_client"})
    with pytest.raises(HTTPException) as e:
        await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret=SECRET_POOL),
                                             decoded=ADMIN)
    assert e.value.status_code == 400 and fs.get(oauth.OAUTH, "pool-1") is None

    discord.token_reply = httpx.Response(200, json={"access_token": "cc-token", "scope": "identify"})
    out = await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret=SECRET_POOL),
                                               decoded=ADMIN)
    assert out["oauth_enabled"] is True
    assert discord.form(1) == {"grant_type": "client_credentials", "scope": "identify"}
    assert discord.basic(1) == f"201:{SECRET_POOL}"
    assert discord.form(2)["token"] == "cc-token"                    # the check's token is revoked
    assert fs.get(oauth.OAUTH, "pool-1")["client_secret"] == SECRET_POOL
    assert "client_secret" not in fs.get("bot_tokens", "pool-1")

    listed = await tokens_router.list_tokens(_=ADMIN)
    everything = json.dumps([listed, out, fs.data.get("audit_log")]) + caplog.text
    assert SECRET_POOL not in everything and "tok-pool-1" not in everything
    row = next(t for t in listed if t["token_id"] == "pool-1")
    assert row["oauth_enabled"] is True and row["server_count"] == 40 and row["server_cap"] == 90


async def test_unreachable_discord_saves_nothing(fs, discord):
    fs.data[oauth.OAUTH].clear()

    def _down(request):
        raise httpx.ConnectError("down", request=request)
    discord.token_reply = _down
    with pytest.raises(HTTPException) as e:
        await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret=SECRET_POOL),
                                             decoded=ADMIN)
    assert e.value.status_code == 503 and fs.get(oauth.OAUTH, "pool-1") is None


async def test_token_list_is_an_allowlist(fs):
    fs.put("bot_tokens", "pool-1", {**fs.get("bot_tokens", "pool-1"), "client_secret": "LEAK", "future": "x"})
    row = next(t for t in await tokens_router.list_tokens(_=ADMIN) if t["token_id"] == "pool-1")
    assert "client_secret" not in row and "future" not in row
    assert row["token"] == "•••ol-1" and row["name"] == "DRB Radio 1"


async def test_clear_and_delete_remove_the_secret(fs):
    await tokens_router.clear_oauth_secret("pool-1", decoded=ADMIN)
    assert fs.get(oauth.OAUTH, "pool-1") is None
    await tokens_router.delete_token("own-a", _=ADMIN)
    assert fs.get(oauth.OAUTH, "own-a") is None


async def test_malformed_secret_is_refused_without_echo(fs, discord):
    fs.data[oauth.OAUTH].clear()
    with pytest.raises(HTTPException) as e:
        await tokens_router.set_oauth_secret("pool-1", tokens_router.OAuthSecret(client_secret="short-secret"),
                                             decoded=ADMIN)
    assert e.value.status_code == 400 and "short-secret" not in str(e.value.detail)
    assert discord.calls == [] and fs.get(oauth.OAUTH, "pool-1") is None