"""User endpoints — /v1/users/*"""

import base64
import logging
import re
import secrets
import time
import uuid
from datetime import datetime, timedelta, timezone

from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel, field_validator
from sqlalchemy import delete, func, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from meshbay_hub import hub_settings, login_throttle, mail
from meshbay_hub.api.deps import get_current_user, require_user_scope
from meshbay_hub.api.middleware import limiter
from meshbay_hub.api.netutil import client_ip
from meshbay_hub.auth import (
    current_pw_version,
    decrypt_email,
    encrypt_email,
    generate_refresh_token,
    hash_email_blind,
    hash_password,
    hash_refresh_token,
    issue_access_token,
    pw_needs_rehash,
    verify_password,
)
from meshbay_hub.config import HubConfig
from meshbay_hub.db.engine import get_db
from meshbay_hub.db.models import (
    EmailVerification, Group, GroupMember, IPLog, Node, Notification,
    RefreshToken, SwarmSource, User, UserDevice, UserPreference,
)

log = logging.getLogger(__name__)

router = APIRouter(prefix="/v1/users", tags=["users"])

_cfg: HubConfig | None = None

def set_config(cfg: HubConfig) -> None:
    global _cfg
    _cfg = cfg

def _ttl() -> int:
    return _cfg.jwt.access_token_ttl if _cfg else 3600

def _refresh_ttl() -> int:
    return _cfg.jwt.refresh_token_ttl if _cfg else 86400 * 30

VERIFICATION_TTL = 86400  # 24 hours
VERIFICATION_MAX_ATTEMPTS = 10


def _generate_code() -> str:
    return f"{secrets.randbelow(1_000_000):06d}"


async def _verify_captcha_or_raise(token: str | None, request: Request) -> None:
    if not token:
        raise HTTPException(status_code=400, detail="captcha_required")
    from meshbay_hub.captcha import verify_captcha
    ok = await verify_captcha(
        _cfg.captcha.secret_key,  # type: ignore[union-attr]
        token,
        request.client.host if request.client else None,
        _cfg.captcha.host_check,  # type: ignore[union-attr]
        _cfg.captcha.allow_unattributed_host,  # type: ignore[union-attr]
    )
    if not ok:
        raise HTTPException(status_code=400, detail="captcha_failed")


# ── Models ────────────────────────────────────────────────────────────────────

class RegisterRequest(BaseModel):
    username:        str
    email:           str
    password:        str | None = None    # deprecated — legacy native clients
    auth_key:        str | None = None    # PBKDF2-derived, new clients
    # Client-generated account recovery key (docs/auth-confirm.md §4.4). Pure
    # pass-through: appended to the verification e-mail so the user's mailbox
    # backs it up, then dropped. Never written to any table, never logged.
    recovery_key:    str | None = None
    captcha_token:   str | None = None

    @field_validator("username")
    @classmethod
    def username_valid(cls, v: str) -> str:
        v = v.strip()
        if len(v) < 3 or len(v) > 64:
            raise ValueError("username must be 3-64 chars")
        if not v.replace("_", "").replace("-", "").replace(".", "").isalnum():
            raise ValueError("username: only letters, digits, -, _, .")
        return v

    @field_validator("email")
    @classmethod
    def email_valid(cls, v: str) -> str:
        """
        Sanity-check the address (L6): the field was plain `str`, so any junk was
        accepted and stored encrypted forever. Deliberately not RFC 5322 — full
        validation would pull in the email-validator dependency for little gain,
        and the address is only ever used for recovery and legal contact.
        """
        v = v.strip()
        local, sep, domain = v.partition("@")
        if (not sep or not local or not domain
                or "." not in domain
                or len(v) > 254
                or any(c.isspace() or ord(c) < 32 for c in v)):
            raise ValueError("invalid email address")
        return v


class LoginRequest(BaseModel):
    username: str
    password: str | None = None    # legacy (raw password) for migration
    auth_key: str | None = None    # PBKDF2-derived auth key (new scheme)


class RefreshRequest(BaseModel):
    refresh_token: str


# ── Endpoints ─────────────────────────────────────────────────────────────────

@router.post("/register", status_code=201)
@limiter.limit("5/minute")
async def register(
    body: RegisterRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    eh = hash_email_blind(body.email)

    existing = await db.execute(
        select(User).where(User.username == body.username))
    found = existing.scalar_one_or_none()

    if found:
        if found.status == "pending" and found.email_hash == eh:
            # Same person retrying before validation — resend a code.
            # No captcha: the initial registration already passed it.
            #
            # Which made this the widest of the three mail doors: no token, no
            # captcha, and the username and address are the caller's own from
            # a moment ago. Registering a victim's address once bought the
            # right to mail them at the endpoint's rate limit for ever.
            # `mail.py` bounds the recipient regardless, and this stops the
            # door being hammered — the answer is the same either way, which
            # is the one this endpoint has always given.
            resend_cooldown = await hub_settings.get_int(
                db, "mail.verification_resend_cooldown",
                hub_settings.mail_default("verification_resend_cooldown"))
            recent = await db.execute(
                select(EmailVerification).where(
                    EmailVerification.user_id == found.id,
                    EmailVerification.purpose == "registration",
                    EmailVerification.created_at
                    > datetime.now(timezone.utc)
                    - timedelta(seconds=resend_cooldown),
                ))
            if not recent.first():
                await _create_and_send_verification(
                    db, found, body.email, eh, body.recovery_key)
                await db.commit()
            return {"user_id": found.id, "email_verification_required": True}
        raise HTTPException(status_code=409, detail="Username already taken")

    # Captcha gate — every fresh registration when a captcha is configured, with
    # no client carve-out. The earlier `and not body.auth_key` exempted anything
    # that sent an `auth_key`, which is *every* real client (the browser sends it
    # too, from the password split) — so the check was off for everyone, and a
    # bot skipped it by sending the field. The desktop client is Chromium and
    # renders the same widget, so it has no need of an exemption either.
    if _cfg and _cfg.captcha.enabled:
        await _verify_captcha_or_raise(body.captcha_token, request)

    # Email uniqueness (only active or pending accounts)
    dup = await db.execute(
        select(User).where(User.email_hash == eh, User.status.in_(["active", "pending"])))
    if dup.scalar_one_or_none():
        raise HTTPException(status_code=409, detail="Email already in use")

    credential = body.auth_key or body.password
    if not credential:
        raise HTTPException(status_code=400, detail="auth_key or password required")

    pw_hash, pw_salt = hash_password(credential)
    pw_ver = current_pw_version() if body.auth_key else 2
    hub_id = _cfg.identity.id if _cfg else "meshbay.org"
    user = User(
        username=body.username,
        email=encrypt_email(body.email),
        email_hash=eh,
        pw_hash=pw_hash,
        pw_salt=pw_salt,
        pw_version=pw_ver,
        hub_id=hub_id,
    )
    db.add(user)
    await db.flush()
    db.add(IPLog(
        user_id=user.id,
        event="account_create",
        ip_address=client_ip(request),
        detail=body.username,
    ))

    await _create_and_send_verification(db, user, body.email, eh, body.recovery_key)
    await db.commit()

    return {"user_id": user.id, "email_verification_required": True}


async def _create_and_send_verification(
    db: AsyncSession, user: User, email: str, eh: str,
    recovery_key: str | None = None,
) -> None:
    user.status = "pending"
    # Invalidate any previous pending verification for this user
    prev = await db.execute(
        select(EmailVerification).where(
            EmailVerification.user_id == user.id,
            EmailVerification.purpose == "registration",
            EmailVerification.verified_at.is_(None),
        ))
    for old in prev.scalars().all():
        await db.delete(old)

    code = _generate_code()
    db.add(EmailVerification(
        email_hash=eh,
        code=code,
        purpose="registration",
        user_id=user.id,
        expires_at=datetime.now(timezone.utc) + timedelta(seconds=VERIFICATION_TTL),
    ))
    await db.flush()
    await mail.send_off_loop(
        db, mail.send_verification_code, email, code,
        purpose="registration", recovery_key=recovery_key)


class VerifyEmailRequest(BaseModel):
    email: str
    code:  str


@router.post("/verify-email")
@limiter.limit("10/minute")
async def verify_email(
    body: VerifyEmailRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    """Verify a registration email with the code received by mail."""
    eh = hash_email_blind(body.email)
    now = datetime.now(timezone.utc)

    result = await db.execute(
        select(EmailVerification).where(
            EmailVerification.email_hash == eh,
            EmailVerification.purpose == "registration",
            EmailVerification.verified_at.is_(None),
        ).order_by(EmailVerification.created_at.desc()))
    verif = result.scalar_one_or_none()

    if not verif:
        raise HTTPException(status_code=404,
                            detail="No pending verification for this email")

    if verif.expires_at.replace(tzinfo=timezone.utc) < now:
        raise HTTPException(status_code=410, detail="Verification code expired")

    if verif.attempts >= VERIFICATION_MAX_ATTEMPTS:
        raise HTTPException(status_code=429, detail="Too many attempts")

    verif.attempts += 1

    if verif.code != body.code.strip():
        await db.commit()
        raise HTTPException(status_code=400, detail="Invalid code")

    verif.verified_at = now
    user = await db.get(User, verif.user_id)
    if user and user.status == "pending":
        user.status = "active"
    await db.commit()
    return {"status": "verified"}


async def _take_login_attempt(db: AsyncSession, username: str) -> None:
    """Spend one passphrase attempt for `username`, or refuse with 429.

    Every path that checks a passphrase goes through here first — `login` and
    `change_password` alike, because a lockout on one door is not a lockout.
    """
    allowed, retry_after = await login_throttle.reserve(db, username)
    if not allowed:
        raise HTTPException(status_code=429, detail="account_locked",
                            headers={"Retry-After": str(retry_after)})


async def _login_failed(db: AsyncSession, username: str, ip: str,
                        user_id: str | None = None) -> None:
    """Record a wrong passphrase and answer 401. Always raises."""
    db.add(IPLog(user_id=user_id, event="login_fail", ip_address=ip, detail=username))
    if await login_throttle.is_now_locked(db, username):
        # Once, on the failure that spent the last attempt — so the logs tab
        # shows when a name was locked, not every refusal after it.
        db.add(IPLog(user_id=user_id, event="login_locked", ip_address=ip,
                     detail=username))
    await db.commit()
    raise HTTPException(status_code=401, detail="Invalid credentials")


@router.post("/login")
@limiter.limit("10/minute")
async def login(
    body: LoginRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    ip = client_ip(request)

    if not body.auth_key and not body.password:
        raise HTTPException(status_code=401, detail="No credentials provided")

    # Before the account is even looked up: an unknown name spends attempts and
    # locks exactly like a real one, so neither answer tells them apart (M1).
    await _take_login_attempt(db, body.username)

    result = await db.execute(
        select(User).where(User.username == body.username))
    user = result.scalar_one_or_none()

    if not user:
        await _login_failed(db, body.username, ip)

    if user.pw_version >= 3:
        # New scheme: verify auth_key
        if not body.auth_key or not verify_password(
            body.auth_key, user.pw_hash, user.pw_salt, version=user.pw_version
        ):
            await _login_failed(db, body.username, ip, user.id)
    else:
        # Legacy scheme: need raw password
        if not body.password:
            # Nothing was checked, so nothing was guessed.
            await login_throttle.release(db, body.username)
            await db.commit()
            raise HTTPException(status_code=401, detail="auth_upgrade_required")
        if not verify_password(
            body.password, user.pw_hash, user.pw_salt, version=user.pw_version
        ):
            await _login_failed(db, body.username, ip, user.id)
        # Migrate to new scheme if auth_key provided alongside password
        if body.auth_key:
            new_hash, new_salt = hash_password(body.auth_key)
            user.pw_hash = new_hash
            user.pw_salt = new_salt
            user.pw_version = current_pw_version()
        elif user.pw_version < 2:
            # Legacy rehash: upgrade Argon2 params within the password scheme (v1 -> v2)
            new_hash, new_salt = hash_password(body.password)
            user.pw_hash = new_hash
            user.pw_salt = new_salt
            user.pw_version = 2

    # The passphrase was right, whatever the account's status turns out to be.
    await login_throttle.clear(db, body.username)

    if user.status != "active":
        await db.commit()
    if user.status == "pending":
        raise HTTPException(status_code=403, detail="email_verification_required")
    if user.status != "active":
        raise HTTPException(status_code=403, detail=f"Account {user.status}")

    # Rehash within the auth_key scheme if Argon2 params upgraded beyond v3
    if user.pw_version >= 3 and pw_needs_rehash(user.pw_version):
        new_hash, new_salt = hash_password(body.auth_key)
        user.pw_hash = new_hash
        user.pw_salt = new_salt
        user.pw_version = current_pw_version()

    memberships = await db.execute(
        select(GroupMember.group_id).where(GroupMember.user_id == user.id))
    group_ids = [gid for (gid,) in memberships.all()]
    access_token = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
    raw_rt, rt_hash = generate_refresh_token()
    family_id = str(uuid.uuid4())

    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(
        user_id=user.id, token_hash=rt_hash,
        family_id=family_id, expires_at=expires_at,
    ))
    db.add(IPLog(user_id=user.id, event="login", ip_address=ip))
    await db.commit()

    return {
        "access_token":  access_token,
        "refresh_token": raw_rt,
        "token_type":    "bearer",
        "expires_in":    _ttl(),
    }


# ── Device authentication ────────────────────────────────────────────────────
#
# A device signs in with an Ed25519 key instead of re-deriving one from the
# passphrase every time. The passphrase remains the account's credential and its
# only recovery path; this is the day-to-day path once a device is registered.
#
# This is **not** the key directory that was H3, and the difference matters:
# nothing reads these but the hub, no group key is ever wrapped for one, and it
# is a different key from the per-node identity keys, which never leave the
# device-node relationship. What it does cost is metadata — the hub now knows
# how many devices an account has and when each last signed in.

DEVICE_AUTH_TIMESTAMP_WINDOW = 60  # seconds, as for node auth


class DeviceRegisterRequest(BaseModel):
    pk_auth_ed25519: str          # base64 raw 32 bytes
    label: str = ""


class DeviceAuthRequest(BaseModel):
    username:  str
    timestamp: int                # unix epoch seconds
    signature: str                # base64 Ed25519 over the message below


@router.post("/devices", status_code=201)
async def register_device(
    body: DeviceRegisterRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """
    Register a device's hub authentication key.

    Requires an existing session, which in practice means the passphrase was
    entered on this device a moment ago. A device cannot enrol itself.
    """
    try:
        raw = base64.b64decode(body.pk_auth_ed25519)
        Ed25519PublicKey.from_public_bytes(raw)
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid Ed25519 public key")

    existing = await db.execute(
        select(UserDevice).where(
            UserDevice.pk_auth_ed25519 == body.pk_auth_ed25519))
    found = existing.scalar_one_or_none()
    if found:
        if found.user_id != current_user.id:
            # One key, one account. Sharing it would make "who signed in" a
            # question with two answers.
            raise HTTPException(status_code=409,
                                detail="That key belongs to another account")
        return {"id": found.id, "label": found.label, "existing": True}

    count = await db.execute(
        select(UserDevice).where(UserDevice.user_id == current_user.id))
    if len(count.scalars().all()) >= 10:
        raise HTTPException(status_code=409,
                            detail="Too many devices — remove one first")

    device = UserDevice(user_id=current_user.id,
                        pk_auth_ed25519=body.pk_auth_ed25519,
                        label=body.label[:64])
    db.add(device)
    await db.commit()
    await db.refresh(device)
    return {"id": device.id, "label": device.label, "existing": False}


@router.get("/devices")
async def list_devices(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(UserDevice).where(UserDevice.user_id == current_user.id)
        .order_by(UserDevice.created_at))
    return {"devices": [
        {"id": d.id, "label": d.label,
         "created_at": d.created_at.isoformat() if d.created_at else None,
         "last_seen": d.last_seen.isoformat() if d.last_seen else None}
        for d in result.scalars().all()
    ]}


@router.delete("/devices/{device_id}")
async def delete_device(
    device_id: str,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """Retire a device's hub key. Its per-node identities are separate and are
    revoked on each node, which the hub cannot do and should not be able to."""
    result = await db.execute(
        select(UserDevice).where(UserDevice.id == device_id,
                                 UserDevice.user_id == current_user.id))
    device = result.scalar_one_or_none()
    if not device:
        raise HTTPException(status_code=404, detail="No such device")
    await db.delete(device)
    await db.commit()
    return {"status": "deleted", "id": device_id}


@router.post("/auth")
@limiter.limit("10/minute")
async def device_auth(
    body: DeviceAuthRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    """
    Sign in with a registered device key. Same shape as `/v1/nodes/auth`.

    The timestamp window is what stops a captured signature being replayed
    later; the signature covers the username as well, so one collected for a
    different account is not usable here.
    """
    now = int(time.time())
    if abs(now - body.timestamp) > DEVICE_AUTH_TIMESTAMP_WINDOW:
        raise HTTPException(status_code=401,
                            detail="Timestamp too old or too far in the future")

    result = await db.execute(select(User).where(User.username == body.username))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=401, detail="Invalid credentials")
    if user.status == "pending":
        raise HTTPException(status_code=403, detail="email_verification_required")
    if user.status != "active":
        raise HTTPException(status_code=403, detail=f"Account {user.status}")

    devices = await db.execute(
        select(UserDevice).where(UserDevice.user_id == user.id))
    message = f"meshbay:user_auth:{body.username}:{body.timestamp}".encode()
    try:
        sig = base64.b64decode(body.signature)
    except Exception:
        raise HTTPException(status_code=401, detail="Invalid signature encoding")

    matched = None
    for device in devices.scalars().all():
        try:
            pk = Ed25519PublicKey.from_public_bytes(
                base64.b64decode(device.pk_auth_ed25519))
            pk.verify(sig, message)
        except Exception:
            continue
        matched = device
        break

    if matched is None:
        db.add(IPLog(user_id=user.id, event="device_auth_fail",
                     ip_address=client_ip(request), detail=body.username))
        await db.commit()
        raise HTTPException(status_code=401, detail="Invalid signature")

    matched.last_seen = datetime.now(timezone.utc)

    memberships = await db.execute(
        select(GroupMember.group_id).where(GroupMember.user_id == user.id))
    group_ids = [gid for (gid,) in memberships.all()]
    access_token = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
    raw_rt, rt_hash = generate_refresh_token()
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(user_id=user.id, token_hash=rt_hash,
                        family_id=str(uuid.uuid4()), expires_at=expires_at))
    db.add(IPLog(user_id=user.id, event="device_auth",
                 ip_address=client_ip(request)))
    await db.commit()

    return {
        "access_token":  access_token,
        "refresh_token": raw_rt,
        "token_type":    "bearer",
        "expires_in":    _ttl(),
        "device_id":     matched.id,
    }


@router.post("/token/refresh")
@limiter.limit("20/minute")
async def token_refresh(
    body: RefreshRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    rt_hash = hash_refresh_token(body.refresh_token)
    result = await db.execute(
        select(RefreshToken).where(RefreshToken.token_hash == rt_hash))
    rt = result.scalar_one_or_none()

    if not rt:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    if rt.revoked:
        # Reuse detected — revoke entire token family
        await db.execute(
            RefreshToken.__table__.update()
            .where(RefreshToken.family_id == rt.family_id)
            .values(revoked=True))
        await db.commit()
        raise HTTPException(status_code=401, detail="Token reuse detected — family revoked")

    if rt.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc):
        raise HTTPException(status_code=401, detail="Expired refresh token")

    user = await db.get(User, rt.user_id)
    if not user or user.status != "active":
        raise HTTPException(status_code=401, detail="User not found or suspended")

    # Revoke old token
    rt.revoked = True

    # Issue new refresh token in the same family
    new_raw_rt, new_rt_hash = generate_refresh_token()
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(
        user_id=user.id, token_hash=new_rt_hash,
        family_id=rt.family_id, expires_at=expires_at,
    ))

    memberships = await db.execute(
        select(GroupMember.group_id).where(GroupMember.user_id == user.id))
    group_ids = [gid for (gid,) in memberships.all()]
    new_access = issue_access_token(user.id, ttl=_ttl(), groups=group_ids)
    await db.commit()

    return {
        "access_token": new_access,
        "refresh_token": new_raw_rt,
        "token_type": "bearer",
        "expires_in": _ttl(),
    }


@router.get("/me")
async def get_current_user_info(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    email = ""
    try:
        email = decrypt_email(current_user.email) if current_user.email else ""
    except Exception:
        pass
    return {
        "user_id":  current_user.id,
        "username": current_user.username,
        "email":    email,
        "role":     current_user.role,
        "status":   current_user.status,
        # Seconds left on a sign-in lockout, 0 when there is none. Told to the
        # account's own session only, so it reveals nothing about anyone else.
        # A passphrase change re-wraps every node's bundle *before* the hub
        # accepts the new passphrase, and must not start while the hub would
        # then refuse it.
        "passphrase_locked_for": await login_throttle.locked_for(db, current_user.username),
    }


class UpdateProfileRequest(BaseModel):
    email: str | None = None

    @field_validator("email")
    @classmethod
    def email_valid(cls, v: str | None) -> str | None:
        if v is None:
            return v
        v = v.strip()
        local, sep, domain = v.partition("@")
        if (not sep or not local or not domain
                or "." not in domain
                or len(v) > 254
                or any(c.isspace() or ord(c) < 32 for c in v)):
            raise ValueError("invalid email address")
        return v


@router.patch("/me")
@limiter.limit("10/minute")
async def update_profile(
    body: UpdateProfileRequest,
    request: Request,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """Update the signed-in account. Changing the address sends a code to it.

    Availability: this is the third path that makes the hub send mail, and it
    was the one with no rate limit and no captcha — while `register` and
    `password/reset-request` have both. The address is any string the caller
    types, and the duplicate check below only rejects one already held by an
    account here, so every address that is *not* registered on this hub was a
    valid target. Any signed-in user could therefore have the hub mail
    arbitrary strangers, from its own domain, as fast as it would go: a relay
    for verification codes with someone else's reputation attached.
    """
    pending_email = None
    if body.email is not None:
        new_email = body.email.strip()
        eh = hash_email_blind(new_email)

        # How often one account may point the hub at a *different* address.
        # Long, because this is the only path where a signed-in account chooses
        # who receives a message, and a short delay alone still allows one
        # stranger per interval indefinitely.
        #
        # Asking again for the address already pending is exempt: it reaches no
        # new recipient — and that recipient is bounded by the per-destination
        # allowance anyway — while without the exemption a typo would lock the
        # account out of correcting it for the whole window.
        pending = (await db.execute(
            select(EmailVerification).where(
                EmailVerification.user_id == current_user.id,
                EmailVerification.purpose == "email_change",
                EmailVerification.verified_at.is_(None),
            ).order_by(EmailVerification.created_at.desc()))).scalars().first()
        same_address_again = pending is not None and pending.email_hash == eh

        if not same_address_again:
            cooldown = await hub_settings.get_int(
                db, "mail.email_change_cooldown",
                hub_settings.mail_default("email_change_cooldown"))
            since = datetime.now(timezone.utc) - timedelta(seconds=cooldown)
            recent = await db.execute(
                select(IPLog).where(
                    IPLog.user_id == current_user.id,
                    IPLog.event == "email_change_request",
                    IPLog.timestamp > since,
                ))
            if recent.first():
                raise HTTPException(
                    status_code=429,
                    detail="This account changed its address recently. "
                           "Try again later, or ask for a new code for the "
                           "address already pending.")

        # Check that no other active/pending account uses this email
        dup = await db.execute(
            select(User).where(
                User.email_hash == eh,
                User.id != current_user.id,
                User.status.in_(["active", "pending"]),
            ))
        if dup.scalar_one_or_none():
            raise HTTPException(status_code=409, detail="Email already in use")

        # Invalidate previous pending email_change verifications for this user
        prev = await db.execute(
            select(EmailVerification).where(
                EmailVerification.user_id == current_user.id,
                EmailVerification.purpose == "email_change",
                EmailVerification.verified_at.is_(None),
            ))
        for old in prev.scalars().all():
            await db.delete(old)

        code = _generate_code()
        db.add(EmailVerification(
            email_hash=eh,
            email_encrypted=encrypt_email(new_email),
            code=code,
            purpose="email_change",
            user_id=current_user.id,
            expires_at=datetime.now(timezone.utc) + timedelta(seconds=VERIFICATION_TTL),
        ))
        db.add(IPLog(user_id=current_user.id, event="email_change_request",
                     ip_address=client_ip(request)))
        await db.flush()
        await mail.send_off_loop(db, mail.send_email_change_code, new_email, code,
                                 purpose="email_change")
        pending_email = new_email

    await db.commit()
    email = ""
    try:
        email = decrypt_email(current_user.email) if current_user.email else ""
    except Exception:
        pass
    resp = {
        "user_id":  current_user.id,
        "username": current_user.username,
        "email":    email,
        "role":     current_user.role,
        "status":   current_user.status,
    }
    if pending_email:
        resp["email_verification_required"] = True
    return resp


class VerifyEmailChangeRequest(BaseModel):
    code: str


@router.post("/verify-email-change")
@limiter.limit("10/minute")
async def verify_email_change(
    body: VerifyEmailChangeRequest,
    request: Request,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """Confirm an email change with the code sent to the new address."""
    now = datetime.now(timezone.utc)

    result = await db.execute(
        select(EmailVerification).where(
            EmailVerification.user_id == current_user.id,
            EmailVerification.purpose == "email_change",
            EmailVerification.verified_at.is_(None),
        ).order_by(EmailVerification.created_at.desc()))
    verif = result.scalar_one_or_none()

    if not verif:
        raise HTTPException(status_code=404,
                            detail="No pending email change")

    if verif.expires_at.replace(tzinfo=timezone.utc) < now:
        raise HTTPException(status_code=410, detail="Verification code expired")

    if verif.attempts >= VERIFICATION_MAX_ATTEMPTS:
        raise HTTPException(status_code=429, detail="Too many attempts")

    verif.attempts += 1

    if verif.code != body.code.strip():
        await db.commit()
        raise HTTPException(status_code=400, detail="Invalid code")

    verif.verified_at = now
    current_user.email = verif.email_encrypted
    current_user.email_hash = verif.email_hash
    await db.commit()

    email = ""
    try:
        email = decrypt_email(current_user.email) if current_user.email else ""
    except Exception:
        pass
    return {"status": "verified", "email": email}


# ── Passphrase change (Flow A) ──────────────────────────────────────────────
#
# docs/auth-confirm.md §3. The passphrase derives two independent values on the
# client: auth_key (verified here) and bundle_key (AES-GCM key for the per-node
# identity bundles, which live on nodes and never on the hub). The client
# re-wraps those bundles from the old bundle_key to the new one on every
# reachable node *before* calling this; the hub only swaps its auth_key
# verifier. There is no email round-trip — the current passphrase is the second
# factor, exactly as for account deletion.


class ChangePasswordRequest(BaseModel):
    old_auth_key: str
    new_auth_key: str


@router.post("/password")
@limiter.limit("5/minute")
async def change_password(
    body: ChangePasswordRequest,
    request: Request,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    await _take_login_attempt(db, current_user.username)
    if not verify_password(body.old_auth_key, current_user.pw_hash,
                           current_user.pw_salt, current_user.pw_version):
        raise HTTPException(status_code=403,
                            detail="Current passphrase does not match")
    await login_throttle.clear(db, current_user.username)
    if body.new_auth_key == body.old_auth_key:
        raise HTTPException(status_code=400,
                            detail="New passphrase must differ from the current one")

    new_hash, new_salt = hash_password(body.new_auth_key)
    current_user.pw_hash = new_hash
    current_user.pw_salt = new_salt
    current_user.pw_version = current_pw_version()

    # Every existing session goes stale. Revoke all refresh tokens, then hand
    # this caller a fresh pair so the tab that made the change stays signed in;
    # other browsers fail their next renewal and drop to the sign-in form.
    await db.execute(
        update(RefreshToken)
        .where(RefreshToken.user_id == current_user.id)
        .values(revoked=True))

    memberships = await db.execute(
        select(GroupMember.group_id).where(GroupMember.user_id == current_user.id))
    group_ids = [gid for (gid,) in memberships.all()]
    access_token = issue_access_token(current_user.id, ttl=_ttl(), groups=group_ids)
    raw_rt, rt_hash = generate_refresh_token()
    expires_at = datetime.now(timezone.utc) + timedelta(seconds=_refresh_ttl())
    db.add(RefreshToken(
        user_id=current_user.id, token_hash=rt_hash,
        family_id=str(uuid.uuid4()), expires_at=expires_at,
    ))
    db.add(IPLog(user_id=current_user.id, event="password_change",
                 ip_address=client_ip(request)))
    await db.commit()

    return {
        "status":        "changed",
        "access_token":  access_token,
        "refresh_token": raw_rt,
        "token_type":    "bearer",
        "expires_in":    _ttl(),
    }


# ── Passphrase reset (Flow B) ──────────────────────────────────────────────
#
# docs/auth-confirm.md §4.2. An e-mail code re-opens hub login for someone who
# has lost their passphrase. It recovers no group content — that needs the
# recovery key, which the client applies on its own after the reset.
# reset-request never reveals whether an account exists.

PASSWORD_RESET_TTL = 3600  # 1 hour — shorter than sign-up verification


class ResetRequestRequest(BaseModel):
    username:      str
    email:         str    # must match the address on file for `username`
    captcha_token: str | None = None

    @field_validator("email")
    @classmethod
    def email_shape(cls, v: str) -> str:
        v = v.strip()
        local, sep, domain = v.partition("@")
        if (not sep or not local or not domain
                or "." not in domain
                or len(v) > 254
                or any(c.isspace() or ord(c) < 32 for c in v)):
            raise ValueError("invalid email address")
        return v


class ResetPasswordRequest(BaseModel):
    username: str
    code: str
    new_auth_key: str


@router.post("/password/reset-request")
@limiter.limit("5/minute")
async def password_reset_request(
    body: ResetRequestRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    if _cfg and _cfg.captcha.enabled:
        await _verify_captcha_or_raise(body.captcha_token, request)

    result = await db.execute(select(User).where(User.username == body.username))
    user = result.scalar_one_or_none()

    # The username and the e-mail must be the pair on file. A mismatch is
    # answered exactly like an unknown account — no reset code is created, no
    # mail is sent — so this reveals nothing and cannot be used to spray reset
    # mail at someone by knowing only their username.
    matched = (
        user is not None
        and user.status == "active"
        and user.email_hash == hash_email_blind(body.email)
    )

    if matched:
        # Per account, under the per-IP limit above. Knowing the pair is the
        # hard part and this endpoint is careful about it, but once someone
        # does, the cost of repeating lands in a mailbox that is not theirs.
        reset_cooldown = await hub_settings.get_int(
            db, "mail.reset_cooldown",
            hub_settings.mail_default("reset_cooldown"))
        recent = await db.execute(
            select(EmailVerification).where(
                EmailVerification.user_id == user.id,
                EmailVerification.purpose == "password_reset",
                EmailVerification.created_at
                > datetime.now(timezone.utc) - timedelta(seconds=reset_cooldown),
            ))
        if recent.first():
            return {"status": "sent_if_exists"}

        prev = await db.execute(
            select(EmailVerification).where(
                EmailVerification.user_id == user.id,
                EmailVerification.purpose == "password_reset",
                EmailVerification.verified_at.is_(None),
            ))
        for old in prev.scalars().all():
            await db.delete(old)

        code = _generate_code()
        db.add(EmailVerification(
            email_hash=user.email_hash or "",
            code=code,
            purpose="password_reset",
            user_id=user.id,
            expires_at=datetime.now(timezone.utc)
            + timedelta(seconds=PASSWORD_RESET_TTL),
        ))
        db.add(IPLog(user_id=user.id, event="password_reset_request",
                     ip_address=client_ip(request)))
        await db.flush()
        try:
            await mail.send_off_loop(
                db, mail.send_password_reset_code, decrypt_email(user.email), code,
                purpose="password_reset")
        except Exception:
            log.exception("Failed to send passphrase reset code")
        await db.commit()
    else:
        db.add(IPLog(
            user_id=user.id if user else None,
            event="password_reset_request",
            ip_address=client_ip(request), detail=body.username))
        await db.commit()

    # Same answer whether or not the username/e-mail pair matched an account.
    return {"status": "sent_if_exists"}


@router.post("/password/reset")
@limiter.limit("10/minute")
async def password_reset(
    body: ResetPasswordRequest,
    request: Request,
    db: AsyncSession = Depends(get_db),
):
    now = datetime.now(timezone.utc)
    result = await db.execute(select(User).where(User.username == body.username))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(status_code=400, detail="Invalid code")

    vres = await db.execute(
        select(EmailVerification).where(
            EmailVerification.user_id == user.id,
            EmailVerification.purpose == "password_reset",
            EmailVerification.verified_at.is_(None),
        ).order_by(EmailVerification.created_at.desc()))
    verif = vres.scalar_one_or_none()
    if not verif:
        raise HTTPException(status_code=404,
                            detail="No pending reset for this account")
    if verif.expires_at.replace(tzinfo=timezone.utc) < now:
        raise HTTPException(status_code=410, detail="Reset code expired")
    if verif.attempts >= VERIFICATION_MAX_ATTEMPTS:
        raise HTTPException(status_code=429, detail="Too many attempts")

    verif.attempts += 1
    if verif.code != body.code.strip():
        await db.commit()
        raise HTTPException(status_code=400, detail="Invalid code")

    verif.verified_at = now
    new_hash, new_salt = hash_password(body.new_auth_key)
    user.pw_hash = new_hash
    user.pw_salt = new_salt
    user.pw_version = current_pw_version()

    # A lost passphrase is a "control may be lost" event: every session dies and
    # every registered device key is dropped, so a stored one cannot sign back
    # in past the reset. Each device re-enrols with the new passphrase.
    await db.execute(
        update(RefreshToken).where(RefreshToken.user_id == user.id)
        .values(revoked=True))
    await db.execute(delete(UserDevice).where(UserDevice.user_id == user.id))
    # A code sent to the address on file is a stronger proof than a passphrase,
    # and it is the way out of a lockout somebody else caused.
    await login_throttle.clear(db, user.username)
    db.add(IPLog(user_id=user.id, event="password_reset",
                 ip_address=client_ip(request)))
    await db.commit()

    # The client already holds new_auth_key; it signs in through the normal
    # path next, which issues tokens and the membership claim.
    return {"status": "reset"}


# ── User preferences ────────────────────────────────────────────────────────

ALLOWED_PREF_KEYS = frozenset([
    "notifications_disabled",
    "default_tab",
    "music_keep_screen_on",
])

# `default_tab:<group_id>`, which is what the SPA writes (group-page.js). The
# suffix used to be unchecked, and the route is `{key:path}`, so any string of
# any length was a distinct key: one account could write unbounded rows into a
# table shared with everyone, each with an unbounded `value` (the column is
# Text). A key over 64 characters was also not a 400 but a 500 — the column is
# String(64), which PostgreSQL enforces and SQLite does not, so it would have
# appeared in production and in no test.
_PREF_GROUP_KEY = re.compile(
    r"^default_tab:[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}"
    r"-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$")

# A person cannot be in more groups than this on one hub without noticing; the
# bound is on rows, because that is what is shared.
MAX_PREFERENCES_PER_ACCOUNT = 500
MAX_PREFERENCE_VALUE = 256


def _valid_pref_key(key: str) -> bool:
    if key in ALLOWED_PREF_KEYS:
        return True
    return bool(_PREF_GROUP_KEY.match(key))


@router.get("/me/preferences")
async def get_preferences(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(UserPreference).where(UserPreference.user_id == current_user.id))
    prefs = {p.key: p.value for p in result.scalars().all()}
    return prefs


class PrefValue(BaseModel):
    value: str


@router.put("/me/preferences/{key:path}")
async def set_preference(
    key: str,
    body: PrefValue,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    if not _valid_pref_key(key):
        raise HTTPException(status_code=400, detail=f"Unknown preference key: {key[:80]}")
    if len(body.value) > MAX_PREFERENCE_VALUE:
        raise HTTPException(status_code=422, detail="Preference value too long")
    result = await db.execute(
        select(UserPreference).where(
            UserPreference.user_id == current_user.id,
            UserPreference.key == key))
    pref = result.scalar_one_or_none()
    if pref:
        pref.value = body.value
    else:
        held = (await db.execute(
            select(func.count()).select_from(UserPreference)
            .where(UserPreference.user_id == current_user.id))).scalar() or 0
        if held >= MAX_PREFERENCES_PER_ACCOUNT:
            raise HTTPException(
                status_code=429, detail="Too many stored preferences")
        db.add(UserPreference(
            user_id=current_user.id, key=key, value=body.value))
    await db.commit()
    return {"key": key, "value": body.value}


@router.delete("/me/preferences/{key:path}")
async def delete_preference(
    key: str,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(UserPreference).where(
            UserPreference.user_id == current_user.id,
            UserPreference.key == key))
    pref = result.scalar_one_or_none()
    if pref:
        await db.delete(pref)
        await db.commit()
    return {"status": "deleted", "key": key}


class NodeKeyRequest(BaseModel):
    pk_node_ed25519: str   # base64 raw 32B Ed25519 public key


@router.put("/me/node_key")
async def register_node_key(
    body: NodeKeyRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """Link a node daemon's Ed25519 public key to the operator's account."""
    try:
        raw = base64.b64decode(body.pk_node_ed25519)
        if len(raw) != 32:
            raise ValueError
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid Ed25519 public key (need 32 bytes base64)")

    current_user.pk_node_ed25519 = body.pk_node_ed25519
    await db.commit()
    return {"status": "stored", "pk_node_ed25519": body.pk_node_ed25519}


@router.delete("/me/node_key")
async def unlink_node_key(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Remove the linked node key from the operator's account."""
    current_user.pk_node_ed25519 = None
    await db.commit()
    return {"status": "unlinked"}


# Key rotation used to live here (`PUT /me/keys`). Identity keys are per node
# now, so rotating means `meshbay-node member unpin <user>` and pairing again with
# a fresh code — an operator decision on the machine that pinned it, not a hub
# call that silently changes what every node believes about someone.


# ── Account deletion ─────────────────────────────────────────────────────────

async def erase_account(db: AsyncSession, user: User, owned_groups: str = "refuse") -> dict:
    """
    Erase an account, keeping only what the law asked us to keep.

    Groups the account owns: `"refuse"` (the owner's own deletion) answers 409
    with their names, because deleting them strands their members and the
    owner can hand them over first. `"delete"` (an administrator's) deletes
    them with the account — an erasure an authority has ordered cannot wait on
    the person it is about.

    Gone: credentials, email, node key, group memberships, notifications, refresh
    tokens, node registrations, device keys, public-swarm sources. The username
    is released.

    Device keys go even though the desktop client keeps its private half: left
    behind, the key still belongs to this tombstone, so an account created later
    from the same installation is refused that device ("belongs to another
    account"). Swarm sources are keyed by the *user* id and carry the node's
    ip:port.

    Kept: the row itself, emptied, and the IP log that points at it. Those logs
    exist for one year to answer legal requests, and a log that cannot say whose
    connection it recorded does not do that — detaching them would keep the data
    and lose the only thing it is for. So the account becomes a tombstone rather
    than a hole in the table.

    Not touched, because the hub cannot: files this person uploaded to nodes, and
    the identity keys nodes pinned for them. Those live on machines the hub does
    not command, and only their operators can remove them.
    """
    owned = (await db.execute(
        select(Group).where(Group.admin_id == user.id))).scalars().all()
    deleted_groups = [{"id": g.id, "name": g.name} for g in owned]
    if owned and owned_groups != "delete":
        raise HTTPException(
            status_code=409,
            detail=("This account still owns groups: "
                    + ", ".join(g.name for g in owned)
                    + ". Delete them or hand them over first — deleting the "
                      "account would strand their members."),
        )
    if owned:
        from meshbay_hub.db.purge import purge_groups
        await purge_groups(db, [g["id"] for g in deleted_groups])

    await db.execute(delete(UserPreference).where(UserPreference.user_id == user.id))
    await db.execute(delete(GroupMember).where(GroupMember.user_id == user.id))
    await db.execute(delete(Notification).where(Notification.user_id == user.id))
    await db.execute(delete(RefreshToken).where(RefreshToken.user_id == user.id))
    await db.execute(delete(Node).where(Node.user_id == user.id))
    await db.execute(delete(UserDevice).where(UserDevice.user_id == user.id))
    await db.execute(delete(SwarmSource).where(SwarmSource.node_id == user.id))
    await db.execute(delete(EmailVerification).where(EmailVerification.user_id == user.id))

    username = user.username
    # Before the name is released: the connection log is kept for its legal
    # retention period and has to stay readable, which means saying who this was
    # and not "deleted-3f9a1c". Nothing else keeps it.
    await db.execute(
        update(IPLog).where(IPLog.user_id == user.id).values(username=username))
    user.username = f"deleted-{user.id[:8]}"
    user.email = ""
    user.email_hash = None
    user.pw_hash = b""
    user.pw_salt = b""
    user.pk_node_ed25519 = None
    user.status = "deleted"
    user.role = "user"
    await db.commit()
    log.info("Account erased: %s (%s), %d owned group(s) deleted",
             username, user.id[:8], len(deleted_groups))
    return {"status": "deleted", "username": username, "user_id": user.id,
            "groups_deleted": deleted_groups}


class DeleteAccountRequest(BaseModel):
    auth_key: str


@router.delete("/me")
async def delete_own_account(
    body: DeleteAccountRequest,
    current_user: User = Depends(require_user_scope),
    db: AsyncSession = Depends(get_db),
):
    """
    Erase your own account. The passphrase is re-checked here.

    A live access token is not enough for something irreversible: it may be a
    borrowed laptop or a session left open. Same value as at sign-in, so the hub
    still never sees the passphrase itself.
    """
    await _take_login_attempt(db, current_user.username)
    if not verify_password(body.auth_key, current_user.pw_hash, current_user.pw_salt,
                           current_user.pw_version):
        raise HTTPException(status_code=403, detail="Passphrase does not match")
    await login_throttle.clear(db, current_user.username)
    return await erase_account(db, current_user)


@router.get("/{username}/pubkeys")
async def get_user_pubkeys(
    username: str,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(select(User).where(User.username == username))
    target = result.scalar_one_or_none()
    if not target:
        raise HTTPException(status_code=404, detail="User not found")
    # Account lookup, not a key directory. `user_id` is how a username is resolved
    # for an invitation, and `pk_node_ed25519` is a node's own linking key. The
    # user identity keys this used to return were H3: whoever asked wrapped the
    # group key for whatever came back.
    resp = {
        "user_id":  target.id,
        "username": target.username,
    }
    if target.pk_node_ed25519:
        resp["pk_node_ed25519"] = target.pk_node_ed25519
    return resp