"""
Passphrase change — Flow A of docs/MESHBAY_DESIGN.md §3.6.

The hub's part is small: re-prove the current passphrase, swap the auth_key
verifier, invalidate every other session, keep the caller's. The re-wrapping of
per-node identity bundles is the client's job and does not touch the hub, so it
is not exercised here.
"""

import base64
import hashlib

import pytest
from sqlalchemy import select

from meshbay_hub.db.models import IPLog, RefreshToken, User


def _auth_key(password: str, username: str) -> str:
    salt = hashlib.sha256(f"meshbay:auth:v1:{username}".encode()).digest()
    return base64.b64encode(
        hashlib.pbkdf2_hmac("sha512", password.encode(), salt, 600_000, 32)).decode()


async def _register(client, username, password="the-first-passphrase"):
    r = await client.post("/v1/users/register", json={
        "username": username, "email": f"{username}@example.com",
        "auth_key": _auth_key(password, username),
    })
    assert r.status_code in (200, 201), r.text
    login = await client.post("/v1/users/login", json={
        "username": username, "auth_key": _auth_key(password, username)})
    assert login.status_code == 200, login.text
    return login.json()


@pytest.mark.asyncio
async def test_change_then_sign_in_with_the_new_passphrase(client):
    old, new = "the-first-passphrase", "a-second-passphrase-entirely"
    session = await _register(client, "alice_test", old)

    r = await client.post("/v1/users/password", json={
        "old_auth_key": _auth_key(old, "alice_test"),
        "new_auth_key": _auth_key(new, "alice_test"),
    }, headers={"Authorization": f"Bearer {session['access_token']}"})
    assert r.status_code == 200, r.text
    assert r.json()["status"] == "changed"

    assert (await client.post("/v1/users/login", json={
        "username": "alice_test", "auth_key": _auth_key(old, "alice_test")})).status_code == 401
    assert (await client.post("/v1/users/login", json={
        "username": "alice_test", "auth_key": _auth_key(new, "alice_test")})).status_code == 200


@pytest.mark.asyncio
async def test_wrong_current_passphrase_is_refused_and_changes_nothing(client):
    old = "the-first-passphrase"
    session = await _register(client, "bob_test", old)

    r = await client.post("/v1/users/password", json={
        "old_auth_key": _auth_key("not the passphrase", "bob_test"),
        "new_auth_key": _auth_key("some-new-passphrase", "bob_test"),
    }, headers={"Authorization": f"Bearer {session['access_token']}"})
    assert r.status_code == 403

    assert (await client.post("/v1/users/login", json={
        "username": "bob_test", "auth_key": _auth_key(old, "bob_test")})).status_code == 200


@pytest.mark.asyncio
async def test_new_must_differ_from_old(client):
    old = "the-first-passphrase"
    session = await _register(client, "carol_test", old)

    r = await client.post("/v1/users/password", json={
        "old_auth_key": _auth_key(old, "carol_test"),
        "new_auth_key": _auth_key(old, "carol_test"),
    }, headers={"Authorization": f"Bearer {session['access_token']}"})
    assert r.status_code == 400


@pytest.mark.asyncio
async def test_unauthenticated_call_is_rejected(client):
    """A session is required — the current passphrase alone is not a credential."""
    await _register(client, "dave_test", "the-first-passphrase")
    r = await client.post("/v1/users/password", json={
        "old_auth_key": _auth_key("the-first-passphrase", "dave_test"),
        "new_auth_key": _auth_key("a-new-one", "dave_test"),
    })
    assert r.status_code in (401, 403, 422)


@pytest.mark.asyncio
async def test_other_sessions_are_invalidated_and_the_caller_keeps_one(
        client, db_session):
    old, new = "the-first-passphrase", "a-second-passphrase-entirely"
    first = await _register(client, "erin_test", old)
    # A second browser signs in before the change.
    second = (await client.post("/v1/users/login", json={
        "username": "erin_test", "auth_key": _auth_key(old, "erin_test")})).json()

    r = await client.post("/v1/users/password", json={
        "old_auth_key": _auth_key(old, "erin_test"),
        "new_auth_key": _auth_key(new, "erin_test"),
    }, headers={"Authorization": f"Bearer {first['access_token']}"})
    assert r.status_code == 200, r.text

    # The other browser's refresh token is dead.
    stale = await client.post("/v1/users/token/refresh", json={
        "refresh_token": second["refresh_token"]})
    assert stale.status_code == 401

    # The caller was handed a fresh pair that still works.
    fresh = await client.post("/v1/users/token/refresh", json={
        "refresh_token": r.json()["refresh_token"]})
    assert fresh.status_code == 200, fresh.text

    uid = (await db_session.execute(
        select(User.id).where(User.username == "erin_test"))).scalar_one()
    live = (await db_session.execute(
        select(RefreshToken).where(RefreshToken.user_id == uid,
                                   RefreshToken.revoked.is_(False)))).scalars().all()
    # Only the family issued to the caller (the refresh above rotated it once).
    assert len(live) == 1


@pytest.mark.asyncio
async def test_the_change_is_logged(client, db_session):
    old, new = "the-first-passphrase", "a-second-passphrase-entirely"
    session = await _register(client, "frank_test", old)
    await client.post("/v1/users/password", json={
        "old_auth_key": _auth_key(old, "frank_test"),
        "new_auth_key": _auth_key(new, "frank_test"),
    }, headers={"Authorization": f"Bearer {session['access_token']}"})

    uid = (await db_session.execute(
        select(User.id).where(User.username == "frank_test"))).scalar_one()
    events = {e.event for e in (await db_session.execute(
        select(IPLog).where(IPLog.user_id == uid))).scalars().all()}
    assert "password_change" in events