"""
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 meshbay_hub.db.models import IPLog, RefreshToken, User
from sqlalchemy import select


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