"""Signing up for a Star Haven account, for people without an internet handle."""

import base64
import json
import re
import threading
import time
import uuid
from concurrent.futures import ThreadPoolExecutor
from collections.abc import Iterator
from pathlib import Path
from http.server import BaseHTTPRequestHandler, HTTPServer, ThreadingHTTPServer
from urllib.parse import parse_qs, quote

import pytest
import requests
from conftest import Appview, DevPds
from playwright.sync_api import Page, expect

PASSWORD = "Hunter2hunter2"

# Stands in for Turnstile's widget, which would put a solved challenge's
# response in the form.
TURNSTILE_WIDGET = """
for (const widget of document.querySelectorAll(".cf-turnstile")) {
    const input = document.createElement("input");
    input.type = "hidden";
    input.name = "cf-turnstile-response";
    input.value = "solved";
    widget.append(input);
}
"""


@pytest.fixture
def turnstile_url() -> Iterator[str]:
    """A stand-in for Turnstile's siteverify, failing only the response "fail"."""

    class Handler(BaseHTTPRequestHandler):
        def do_POST(self) -> None:
            length = int(self.headers["Content-Length"])
            form = parse_qs(self.rfile.read(length).decode())
            body = json.dumps({"success": form.get("response") != ["fail"]}).encode()
            self.send_response(200)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def log_message(self, *args) -> None:
            pass

    server = HTTPServer(("127.0.0.1", 0), Handler)
    threading.Thread(target=server.serve_forever, daemon=True).start()
    yield f"http://127.0.0.1:{server.server_port}"
    server.shutdown()


class PdsProxy:
    """Passes requests on to the dev PDS, noting each one's path and
    `X-Forwarded-For`, as the internal address the appview calls it at.

    It also lets an admin account's session mint invite codes, as Tranquil
    PDS does, by trading the session's token for the dev PDS's admin
    password. Tokens it hasn't seen issued, or that `expire_sessions` has
    expired, are refused.
    """

    def __init__(self, pds_url: str) -> None:
        self.forwarded_for: list[tuple[str, str | None]] = []
        self.issued: set[str] = set()
        self.expired: set[str] = set()
        proxy = self

        class Handler(BaseHTTPRequestHandler):
            def relay(self) -> None:
                proxy.forwarded_for.append(
                    (self.path.split("?")[0], self.headers.get("X-Forwarded-For"))
                )
                path = self.path.split("?")[0]
                headers = {
                    name: self.headers[name]
                    for name in ("Content-Type", "Authorization")
                    if name in self.headers
                }
                bearer = headers.get("Authorization", "").removeprefix("Bearer ")
                if path == "/xrpc/com.atproto.server.createInviteCode" and bearer:
                    if bearer in proxy.expired:
                        return self.reply(400, {"error": "ExpiredToken"})
                    if bearer not in proxy.issued:
                        return self.reply(401, {"error": "AuthenticationRequired"})
                    headers["Authorization"] = "Basic " + base64.b64encode(
                        b"admin:admin-pass"
                    ).decode()

                length = int(self.headers.get("Content-Length") or 0)
                response = requests.request(
                    self.command,
                    f"{pds_url}{self.path}",
                    data=self.rfile.read(length),
                    headers=headers,
                    timeout=10,
                )
                if response.ok and path in (
                    "/xrpc/com.atproto.server.createSession",
                    "/xrpc/com.atproto.server.refreshSession",
                ):
                    proxy.issued.add(response.json()["accessJwt"])
                self.send_response(response.status_code)
                self.send_header("Content-Type", response.headers.get("Content-Type", ""))
                self.send_header("Content-Length", str(len(response.content)))
                self.end_headers()
                self.wfile.write(response.content)

            def reply(self, status: int, body: dict) -> None:
                content = json.dumps(body).encode()
                self.send_response(status)
                self.send_header("Content-Type", "application/json")
                self.send_header("Content-Length", str(len(content)))
                self.end_headers()
                self.wfile.write(content)

            do_GET = do_POST = relay

            def log_message(self, *args) -> None:
                pass

        self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
        self.url = f"http://127.0.0.1:{self.server.server_port}"

    def expire_sessions(self) -> None:
        """Makes every session issued so far expire."""
        self.expired |= self.issued

    def sent_for(self, path: str) -> list[str | None]:
        """The `X-Forwarded-For` of every request to `path`."""
        return [header for sent, header in self.forwarded_for if sent == path]


@pytest.fixture
def pds_proxy(dev_pds: DevPds) -> Iterator[PdsProxy]:
    proxy = PdsProxy(dev_pds.pds_url)
    threading.Thread(target=proxy.server.serve_forever, daemon=True).start()
    yield proxy
    proxy.server.shutdown()


@pytest.fixture
def admin_credentials(dev_pds: DevPds, tmp_path: Path) -> Path:
    """A credentials file for an admin account, as pds.starhaven.dev has."""
    handle = f"admin-{uuid.uuid4().hex[:8]}.test"
    requests.post(
        f"{dev_pds.pds_url}/xrpc/com.atproto.server.createAccount",
        json={"handle": handle, "email": f"{handle}@dev.test", "password": PASSWORD},
        timeout=10,
    ).raise_for_status()
    path = tmp_path / "pds-admin.json"
    path.write_text(json.dumps({"identifier": handle, "password": PASSWORD}))
    return path


@pytest.fixture
def server_args(
    turnstile_url: str, pds_proxy: PdsProxy, admin_credentials: Path
) -> list[str]:
    return [
        "--turnstile-verify-url",
        turnstile_url,
        "--signup-pds-internal-url",
        pds_proxy.url,
        "--pds-admin-credentials-file",
        str(admin_credentials),
    ]


def new_name() -> str:
    """A handle's first part, unused on the dev PDS that every test shares."""
    return f"new-{uuid.uuid4().hex[:8]}"


def ask_for_email(
    appview: Appview,
    email: str,
    turnstile: str = "solved",
    headers: dict[str, str] | None = None,
    old_enough: bool = True,
) -> str:
    """Submits the first step of sign-up, returning the page it answers with."""
    response = requests.post(
        f"{appview.url}/signup",
        data={
            "email": email,
            "return_to": "/",
            "cf-turnstile-response": turnstile,
            **({"age": "yes"} if old_enough else {}),
        },
        headers=headers,
        timeout=10,
    )
    response.raise_for_status()
    return response.text


def finish_links(appview: Appview, email: str) -> list[str]:
    """The links in every sign-up email the server has printed for `email`."""
    return re.findall(
        rf"email to {re.escape(email)}\n"
        r"Subject: Finish creating your Star Haven account\n"
        r".*?(http\S+/signup/finish\?token=\S+)",
        appview.log.read_text(),
        re.DOTALL,
    )


def test_signing_up_creates_an_account_and_signs_in(
    page: Page, appview: Appview, dev_pds: DevPds
):
    name = new_name()
    email = f"{name}@dev.test"
    page.route(
        "https://challenges.cloudflare.com/**",
        lambda route: route.fulfill(
            content_type="text/javascript", body=TURNSTILE_WIDGET
        ),
    )

    # The appview stands in for an app listening on this computer.
    app_url = f"{appview.url}/create"
    page.goto(f"{appview.url}/signup?return_to={quote(app_url, safe='')}")
    page.get_by_label("Email address").fill(email)
    page.get_by_label("I'm 13 or older").check()
    page.get_by_role("button", name="Continue").click()
    expect(page.get_by_role("heading", name="Check your email")).to_be_visible()

    [link] = finish_links(appview, email)
    page.goto(link)
    page.get_by_label("Handle").fill(name)
    page.get_by_label("Password").fill(PASSWORD)
    page.get_by_role("button", name="Create account").click()

    # PDS sign in
    page.get_by_role("textbox", name="Password").fill(PASSWORD)
    page.get_by_role("button", name="Sign in").click()
    page.get_by_role("button", name="Authorize").click()

    expect(page).to_have_url(f"{app_url}?handle={name}.test")
    response = requests.get(
        f"{dev_pds.pds_url}/xrpc/com.atproto.identity.resolveHandle",
        params={"handle": f"{name}.test"},
        timeout=5,
    )
    response.raise_for_status()


def test_a_sign_up_link_works_once(appview: Appview):
    name = new_name()
    email = f"{name}@dev.test"
    ask_for_email(appview, email)
    [link] = finish_links(appview, email)
    token = link.split("token=")[1]

    response = requests.post(
        f"{appview.url}/signup/finish",
        data={"token": token, "handle": name, "password": PASSWORD},
        allow_redirects=False,
        timeout=10,
    )
    assert response.is_redirect

    assert "This link has expired" in requests.get(link, timeout=10).text


def test_a_sign_up_link_makes_one_account_however_often_it_is_sent_at_once(
    appview: Appview, dev_pds: DevPds
):
    email = f"{new_name()}@dev.test"
    ask_for_email(appview, email)
    [link] = finish_links(appview, email)
    token = link.split("token=")[1]
    names = [new_name() for _ in range(5)]

    def finish(name: str) -> requests.Response:
        return requests.post(
            f"{appview.url}/signup/finish",
            data={"token": token, "handle": name, "password": PASSWORD},
            allow_redirects=False,
            timeout=30,
        )

    with ThreadPoolExecutor(len(names)) as pool:
        responses = list(pool.map(finish, names))

    assert sum(response.is_redirect for response in responses) == 1
    created = [
        name
        for name in names
        if requests.get(
            f"{dev_pds.pds_url}/xrpc/com.atproto.identity.resolveHandle",
            params={"handle": f"{name}.test"},
            timeout=5,
        ).ok
    ]
    assert len(created) == 1


def test_a_handle_already_taken_asks_for_another(appview: Appview):
    name = new_name()
    for email in [f"{name}@dev.test", f"{name}-again@dev.test"]:
        ask_for_email(appview, email)
    [first] = finish_links(appview, f"{name}@dev.test")
    [second] = finish_links(appview, f"{name}-again@dev.test")

    for link in [first, second]:
        response = requests.post(
            f"{appview.url}/signup/finish",
            data={
                "token": link.split("token=")[1],
                "handle": name,
                "password": PASSWORD,
            },
            allow_redirects=False,
            timeout=10,
        )

    assert f"{name}.test is taken" in response.text


def test_a_failed_security_check_sends_no_email(appview: Appview):
    email = f"{new_name()}@dev.test"

    page = ask_for_email(appview, email, turnstile="fail")

    assert "Complete the security check" in page
    assert finish_links(appview, email) == []


def test_one_address_gets_only_three_emails_a_day(appview: Appview):
    email = f"{new_name()}@dev.test"

    for _ in range(3):
        assert "Check your email" in ask_for_email(appview, email)
    page = ask_for_email(appview, email)

    assert "sent 3 sign-up emails today" in page
    assert len(finish_links(appview, email)) == 3


def test_an_address_with_an_account_gets_a_reminder_instead(
    appview: Appview, dev_pds: DevPds
):
    name = new_name()
    email = f"{name}@dev.test"
    requests.post(
        f"{dev_pds.pds_url}/xrpc/com.atproto.server.createAccount",
        json={"handle": f"{name}.test", "email": email, "password": PASSWORD},
        timeout=10,
    ).raise_for_status()
    newcomer = f"{new_name()}@dev.test"

    page = ask_for_email(appview, email)

    # The same page, but for the address on it, as itself and in the link to
    # send another.
    other = ask_for_email(appview, newcomer)
    other = other.replace(quote(newcomer, safe=""), quote(email, safe=""))
    assert page == other.replace(newcomer, email)
    assert finish_links(appview, email) == []
    assert (
        f"email to {email}\nSubject: You already have a Star Haven account"
        in appview.log.read_text()
    )


def test_the_pds_hears_which_visitor_each_sign_up_is_for(
    appview: Appview, pds_proxy: PdsProxy
):
    name = new_name()
    email = f"{name}@dev.test"
    # As a reverse proxy in front of the appview would say.
    visitor = {"X-Forwarded-For": "203.0.113.7"}

    ask_for_email(appview, email, headers=visitor)
    [link] = finish_links(appview, email)
    response = requests.post(
        f"{appview.url}/signup/finish",
        data={"token": link.split("token=")[1], "handle": name, "password": PASSWORD},
        headers=visitor,
        allow_redirects=False,
        timeout=10,
    )

    assert response.is_redirect
    assert pds_proxy.sent_for("/xrpc/_account.checkEmailInUse") == ["203.0.113.7"]
    assert pds_proxy.sent_for("/xrpc/com.atproto.server.createAccount") == [
        "203.0.113.7"
    ]


def sign_up(appview: Appview) -> None:
    """Signs up a new account, start to finish."""
    name = new_name()
    email = f"{name}@dev.test"
    ask_for_email(appview, email)
    [link] = finish_links(appview, email)
    response = requests.post(
        f"{appview.url}/signup/finish",
        data={"token": link.split("token=")[1], "handle": name, "password": PASSWORD},
        allow_redirects=False,
        timeout=10,
    )
    assert response.is_redirect


def test_the_admin_account_signs_in_once_and_refreshes_its_session(
    appview: Appview, pds_proxy: PdsProxy
):
    sign_up(appview)
    sign_up(appview)
    assert len(pds_proxy.sent_for("/xrpc/com.atproto.server.createSession")) == 1
    assert pds_proxy.sent_for("/xrpc/com.atproto.server.refreshSession") == []

    # The dev PDS's tokens differ only by when they were issued, to the
    # second, so a refresh in the same second would mint one that's expired.
    time.sleep(1.1)
    pds_proxy.expire_sessions()
    sign_up(appview)

    assert len(pds_proxy.sent_for("/xrpc/com.atproto.server.createSession")) == 1
    assert len(pds_proxy.sent_for("/xrpc/com.atproto.server.refreshSession")) == 1


def test_sign_up_needs_someone_13_or_older(appview: Appview):
    email = f"{new_name()}@dev.test"

    page = ask_for_email(appview, email, old_enough=False)

    assert "Confirm you're 13 or older" in page
    assert finish_links(appview, email) == []


def test_the_policies_open_beside_the_sign_up_form(page: Page, appview: Appview):
    page.goto(f"{appview.url}/signup")

    for name in ["terms of service", "privacy policy"]:
        link = page.locator("form").get_by_role("link", name=name)
        expect(link).to_have_attribute("target", "_blank")
        expect(link).to_have_attribute("rel", "noopener")