"""
Passphrase reset by e-mail code — Flow B of docs/auth-confirm.md §4.2.

The hub's part re-opens sign-in only: it swaps the auth_key verifier, kills
every session, and drops every registered device key so a stored one cannot
sign back in past the reset. Restoring group access is the client's job with
the recovery key and is not exercised here.
"""

import base64
import time
from datetime import datetime, timedelta, timezone

import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from meshbay_common.crypto import pk_to_b64
from meshbay_hub.db.models import EmailVerification, IPLog, User
from sqlalchemy import select


def _email(username: str) -> str:
    return f"{username}@example.com"


async def _register(client, username, auth_key="k" * 44):
    r = await client.post("/v1/users/register", json={
        "username": username, "email": _email(username),
        "auth_key": auth_key})
    assert r.status_code in (200, 201), r.text


async def _request_reset(client, username, email=None):
    return await client.post("/v1/users/password/reset-request", json={
        "username": username, "email": email or _email(username)})


async def _reset_code(db_session, username) -> str:
    uid = (await db_session.execute(
        select(User.id).where(User.username == username))).scalar_one()
    row = (await db_session.execute(
        select(EmailVerification).where(
            EmailVerification.user_id == uid,
            EmailVerification.purpose == "password_reset",
            EmailVerification.verified_at.is_(None),
        ).order_by(EmailVerification.created_at.desc()))).scalars().first()
    return row.code if row else None


@pytest.mark.asyncio
async def test_reset_lets_the_user_sign_in_with_a_new_passphrase(client, db_session):
    await _register(client, "alice", "old" + "a" * 41)
    r = await _request_reset(client, "alice")
    assert r.status_code == 200 and r.json()["status"] == "sent_if_exists"

    code = await _reset_code(db_session, "alice")
    assert code

    new = "new" + "b" * 41
    r = await client.post("/v1/users/password/reset", json={
        "username": "alice", "code": code, "new_auth_key": new})
    assert r.status_code == 200, r.text

    assert (await client.post("/v1/users/login", json={
        "username": "alice", "auth_key": "old" + "a" * 41})).status_code == 401
    assert (await client.post("/v1/users/login", json={
        "username": "alice", "auth_key": new})).status_code == 200


@pytest.mark.asyncio
async def test_reset_request_never_reveals_whether_an_account_exists(
        client, db_session):
    r = await _request_reset(client, "ghost")
    assert r.status_code == 200
    assert r.json()["status"] == "sent_if_exists"
    rows = (await db_session.execute(select(EmailVerification))).scalars().all()
    assert rows == []


@pytest.mark.asyncio
async def test_reset_request_needs_the_username_and_email_to_match(client, db_session):
    await _register(client, "hank")

    # Right username, wrong e-mail — answered exactly like an unknown account,
    # and no code is created.
    r = await _request_reset(client, "hank", email="someone.else@example.com")
    assert r.status_code == 200
    assert r.json()["status"] == "sent_if_exists"
    assert (await db_session.execute(
        select(EmailVerification))).scalars().all() == []

    # The real pair does create one.
    await _request_reset(client, "hank")
    assert (await db_session.execute(
        select(EmailVerification))).scalars().first() is not None


@pytest.mark.asyncio
async def test_reset_request_rejects_a_malformed_email(client):
    await _register(client, "iris")
    r = await client.post("/v1/users/password/reset-request", json={
        "username": "iris", "email": "not-an-email"})
    assert r.status_code == 422


@pytest.mark.asyncio
async def test_a_wrong_code_is_rejected_and_counts_against_the_limit(
        client, db_session):
    await _register(client, "bob")
    await _request_reset(client, "bob")

    for _ in range(10):
        r = await client.post("/v1/users/password/reset", json={
            "username": "bob", "code": "000000", "new_auth_key": "x" * 44})
        assert r.status_code == 400
    r = await client.post("/v1/users/password/reset", json={
        "username": "bob", "code": "000000", "new_auth_key": "x" * 44})
    assert r.status_code == 429


@pytest.mark.asyncio
async def test_an_expired_code_is_refused(client, db_session):
    await _register(client, "carol")
    await _request_reset(client, "carol")

    uid = (await db_session.execute(
        select(User.id).where(User.username == "carol"))).scalar_one()
    row = (await db_session.execute(select(EmailVerification).where(
        EmailVerification.user_id == uid))).scalars().one()
    row.expires_at = datetime.now(timezone.utc) - timedelta(minutes=1)
    await db_session.commit()

    r = await client.post("/v1/users/password/reset", json={
        "username": "carol", "code": row.code, "new_auth_key": "y" * 44})
    assert r.status_code == 410


@pytest.mark.asyncio
async def test_a_reset_code_works_once(client, db_session):
    await _register(client, "dave")
    await _request_reset(client, "dave")
    code = await _reset_code(db_session, "dave")

    first = await client.post("/v1/users/password/reset", json={
        "username": "dave", "code": code, "new_auth_key": "z" * 44})
    assert first.status_code == 200
    second = await client.post("/v1/users/password/reset", json={
        "username": "dave", "code": code, "new_auth_key": "z" * 44})
    assert second.status_code == 404


@pytest.mark.asyncio
async def test_reset_revokes_sessions_and_wipes_devices(client, db_session):
    await _register(client, "erin", "erin" + "a" * 40)
    login = await client.post("/v1/users/login", json={
        "username": "erin", "auth_key": "erin" + "a" * 40})
    refresh_token = login.json()["refresh_token"]
    token = login.json()["access_token"]

    sk = Ed25519PrivateKey.generate()
    dev = await client.post(
        "/v1/users/devices",
        json={"pk_auth_ed25519": pk_to_b64(sk.public_key()), "label": "laptop"},
        headers={"Authorization": f"Bearer {token}"})
    assert dev.status_code == 201, dev.text

    await _request_reset(client, "erin")
    code = await _reset_code(db_session, "erin")
    r = await client.post("/v1/users/password/reset", json={
        "username": "erin", "code": code, "new_auth_key": "erin-new" + "b" * 36})
    assert r.status_code == 200

    # Old refresh token is dead.
    assert (await client.post("/v1/users/token/refresh", json={
        "refresh_token": refresh_token})).status_code == 401

    # Every device key is gone; the stored one can no longer sign in.
    uid = (await db_session.execute(
        select(User.id).where(User.username == "erin"))).scalar_one()
    from meshbay_hub.db.models import UserDevice
    devices = (await db_session.execute(
        select(UserDevice).where(UserDevice.user_id == uid))).scalars().all()
    assert devices == []

    ts = int(time.time())
    msg = f"meshbay:user_auth:erin:{ts}".encode()
    da = await client.post("/v1/users/auth", json={
        "username": "erin", "timestamp": ts,
        "signature": base64.b64encode(sk.sign(msg)).decode()})
    assert da.status_code == 401


@pytest.mark.asyncio
async def test_the_request_and_the_reset_are_logged(client, db_session):
    await _register(client, "frank")
    await _request_reset(client, "frank")
    code = await _reset_code(db_session, "frank")
    await client.post("/v1/users/password/reset", json={
        "username": "frank", "code": code, "new_auth_key": "f" * 44})

    uid = (await db_session.execute(
        select(User.id).where(User.username == "frank"))).scalar_one()
    events = {e.event for e in (await db_session.execute(
        select(IPLog).where(IPLog.user_id == uid))).scalars().all()}
    assert {"password_reset_request", "password_reset"} <= events