bambuddy/backend/app/api/routes/users.py

from datetime import datetime, timezone
from typing import Annotated

import jwt as _jwt
from fastapi import APIRouter, Depends, HTTPException, Query, status
from fastapi.security import HTTPAuthorizationCredentials
from sqlalchemy import delete, func, select, update as sql_update
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from backend.app.api.routes.settings import get_external_login_url
from backend.app.core.auth import (
    ALGORITHM,
    SECRET_KEY,
    RequireAdminIfAuthEnabled,
    RequireAnyPermissionIfAuthEnabled,
    RequirePermissionIfAuthEnabled,
    get_current_user_optional,
    get_password_hash,
    revoke_jti,
    security,
    verify_password,
)
from backend.app.core.database import get_db
from backend.app.core.permissions import Permission
from backend.app.core.websocket import ws_manager
from backend.app.models.api_key import APIKey
from backend.app.models.archive import PrintArchive
from backend.app.models.auth_ephemeral import AuthEphemeralToken
from backend.app.models.connected_app import ConnectedApp, ConnectedAppGrant
from backend.app.models.group import Group
from backend.app.models.library import LibraryFile
from backend.app.models.long_lived_token import LongLivedToken
from backend.app.models.oidc_provider import UserOIDCLink
from backend.app.models.print_batch import PrintBatch
from backend.app.models.print_log import PrintLogEntry
from backend.app.models.print_queue import PrintQueueItem
from backend.app.models.settings import Settings
from backend.app.models.user import User
from backend.app.models.user_otp_code import UserOTPCode
from backend.app.models.user_totp import UserTOTP
from backend.app.schemas.auth import (
    ChangePasswordRequest,
    GroupBrief,
    UserCreate,
    UserResponse,
    UserSlim,
    UserUpdate,
)
from backend.app.services.email_service import (
    create_welcome_email_from_template,
    generate_secure_password,
    get_smtp_settings,
    send_email,
)
from backend.app.services.finance_defaults import ensure_user_finance_defaults
from backend.app.services.printer_manager import printer_manager
from backend.app.services.slice_dispatch import slice_dispatch
from backend.app.utils.library_paths import remove_library_photos_dir
from backend.app.utils.local_time import utcnow_naive

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


def _user_to_response(user: User) -> UserResponse:
    """Convert a User model to UserResponse schema."""
    return UserResponse(
        id=user.id,
        username=user.username,
        email=user.email,
        role=user.role,
        is_active=user.is_active,
        is_admin=user.is_admin,
        auth_source=getattr(user, "auth_source", "local"),
        groups=[GroupBrief(id=g.id, name=g.name) for g in user.groups],
        permissions=sorted(user.get_permissions()),
        created_at=user.created_at.isoformat(),
    )


@router.get("", response_model=list[UserResponse])
@router.get("/", response_model=list[UserResponse])
async def list_users(
    _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ),
    db: AsyncSession = Depends(get_db),
):
    """List all users.

    Read-only — gated on ``USERS_READ`` only. Operator-visible UIs
    (Stats filter-by-user, Archives Print Log username column, File
    Manager username autocomplete) consume this endpoint via custom-
    group ``users:read`` grants without admin role. The admin-only
    boundary lives on the write endpoints below."""
    result = await db.execute(select(User).options(selectinload(User.groups)).order_by(User.created_at))
    users = result.scalars().all()
    return [_user_to_response(user) for user in users]


async def drop_pending_tokens(db: AsyncSession, username: str) -> None:
    """Delete the one-time tokens pending under a username that is going away.

    Reset links, 2FA and OIDC steps, connect codes, camera, WebSocket and slicer
    tokens are looked up by username, so they would work for whoever is given
    the name next. The logout list (revoked_jti) stays: dropping an entry would
    make the logged-out token valid again.
    """
    await db.execute(
        delete(AuthEphemeralToken).where(
            # Both sides in SQL: SQLite's lower() changes only ASCII letters, so
            # it never equals Python's lower() of a name with an umlaut.
            func.lower(AuthEphemeralToken.username) == func.lower(username),
            AuthEphemeralToken.token_type != "revoked_jti",
        )
    )


@router.post("", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
@router.post("/", response_model=UserResponse, status_code=status.HTTP_201_CREATED)
async def create_user(
    user_data: UserCreate,
    _admin: User | None = RequireAdminIfAuthEnabled(),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_CREATE),
    db: AsyncSession = Depends(get_db),
):
    """Create a new user.

    When advanced authentication is enabled:
    - Email is required
    - Password is auto-generated and emailed to user
    - Admin cannot set or see the password
    """
    import logging

    logger = logging.getLogger(__name__)

    # Check if advanced auth is enabled
    result = await db.execute(select(Settings).where(Settings.key == "advanced_auth_enabled"))
    advanced_auth_setting = result.scalar_one_or_none()
    advanced_auth_enabled = advanced_auth_setting and advanced_auth_setting.value.lower() == "true"

    # Check if username already exists (case-insensitive)
    existing_user = await db.execute(select(User).where(func.lower(User.username) == func.lower(user_data.username)))
    if existing_user.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Username already exists",
        )

    # Validate role
    if user_data.role not in ["admin", "user"]:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Role must be 'admin' or 'user'",
        )

    # Advanced auth validation
    if advanced_auth_enabled:
        if not user_data.email:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email is required when advanced authentication is enabled",
            )
        # Check if email already exists (case-insensitive)
        existing_email = await db.execute(select(User).where(func.lower(User.email) == func.lower(user_data.email)))
        if existing_email.scalar_one_or_none():
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email already exists",
            )

    # Generate password if advanced auth enabled, otherwise require password
    if advanced_auth_enabled:
        password = generate_secure_password()
    else:
        if not user_data.password:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Password is required when advanced authentication is disabled",
            )
        password = user_data.password

    new_user = User(
        username=user_data.username,
        email=user_data.email,
        password_hash=get_password_hash(password),
        role=user_data.role,
        is_active=True,
    )

    # Handle group assignments
    if user_data.group_ids:
        groups_result = await db.execute(select(Group).where(Group.id.in_(user_data.group_ids)))
        groups = groups_result.scalars().all()
        if len(groups) != len(user_data.group_ids):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="One or more group IDs are invalid",
            )
        new_user.groups = list(groups)

    db.add(new_user)
    await db.flush()
    await ensure_user_finance_defaults(db, new_user)
    await db.commit()
    await db.refresh(new_user)

    # Send welcome email if advanced auth enabled
    if advanced_auth_enabled and new_user.email:
        try:
            smtp_settings = await get_smtp_settings(db)
            if smtp_settings:
                login_url = await get_external_login_url(db)
                subject, text_body, html_body = await create_welcome_email_from_template(
                    db, new_user.username, password, login_url
                )
                send_email(smtp_settings, new_user.email, subject, text_body, html_body)
                logger.info(f"Welcome email sent to {new_user.email}")
            else:
                logger.warning(f"SMTP not configured, could not send welcome email to {new_user.email}")
        except Exception as e:
            logger.error(f"Failed to send welcome email: {e}")
            # Don't fail user creation if email fails

    return _user_to_response(new_user)


@router.get("/slim", response_model=list[UserSlim])
async def list_users_slim(
    _: User | None = RequireAnyPermissionIfAuthEnabled(Permission.USERS_READ_SLIM, Permission.USERS_READ),
    db: AsyncSession = Depends(get_db),
):
    """List users as ``{id, username}`` only (#1894).

    Exists so an API key -- or a group that should not see emails, roles and
    permission sets -- can turn the ``created_by_id`` values it already gets
    back from archives, stats and the queue into names.

    ``USERS_READ`` is accepted alongside ``USERS_READ_SLIM`` because it is
    strictly broader; groups that already hold it keep working without a
    permission backfill. For API keys only the slim permission resolves (the
    full one is unmapped = administrative), so a key reaches this and not the
    listing above.

    Declared before ``/{user_id}`` on purpose: FastAPI matches in declaration
    order, and the reverse order would parse "slim" as the int path parameter
    and answer 422.
    """
    result = await db.execute(select(User.id, User.username).order_by(User.username))
    return [UserSlim(id=row.id, username=row.username) for row in result.all()]


@router.get("/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ),
    db: AsyncSession = Depends(get_db),
):
    """Get a user by ID. Read-only — gated on ``USERS_READ`` only."""
    result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    return _user_to_response(user)


@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
    user_id: int,
    user_data: UserUpdate,
    _admin: User | None = RequireAdminIfAuthEnabled(),
    _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_UPDATE),
    db: AsyncSession = Depends(get_db),
):
    """Update a user."""
    close_sockets = False
    deactivated_keys: list[int] = []
    result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    # Prevent deactivating the last admin
    if user_data.is_active is False and user.is_admin:
        # Count admins by role or Administrators group membership
        admin_count_result = await db.execute(select(User).where(User.role == "admin", User.is_active.is_(True)))
        role_admins = admin_count_result.scalars().all()

        # Also check for users in Administrators group
        admin_group_result = await db.execute(
            select(Group).where(Group.name == "Administrators").options(selectinload(Group.users))
        )
        admin_group = admin_group_result.scalar_one_or_none()
        group_admins = [u for u in (admin_group.users if admin_group else []) if u.is_active]

        # Combine unique admins
        all_admins = {u.id for u in role_admins} | {u.id for u in group_admins}
        if len(all_admins) <= 1 and user.id in all_admins:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Cannot deactivate the last admin user",
            )

    # Prevent changing role of last admin
    if user_data.role and user_data.role != "admin" and user.role == "admin":
        admin_count_result = await db.execute(select(User).where(User.role == "admin", User.is_active.is_(True)))
        admin_count = len(admin_count_result.scalars().all())
        if admin_count <= 1:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Cannot change role of the last admin user",
            )

    if user_data.username is not None:
        # Check if new username already exists (case-insensitive)
        existing_user = await db.execute(
            select(User).where(func.lower(User.username) == func.lower(user_data.username), User.id != user_id)
        )
        if existing_user.scalar_one_or_none():
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Username already exists",
            )
        if user_data.username != user.username:
            # The name passes on: a deleted user's still-valid JWTs under the new
            # name must not sign this user in, and what is pending under the old
            # one must not reach whoever is given it next. This user's own
            # sessions carry the old name and end anyway. A change of case counts
            # too: SQLite compares only ASCII letters without case, so 'Ärger'
            # and 'ärger' are two names there.
            await drop_pending_tokens(db, user.username)
            user.password_changed_at = utcnow_naive()
            close_sockets = True
        user.username = user_data.username

    if user_data.email is not None:
        # Check if new email already exists (case-insensitive)
        existing_email = await db.execute(
            select(User).where(func.lower(User.email) == func.lower(user_data.email), User.id != user_id)
        )
        if existing_email.scalar_one_or_none():
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Email already exists",
            )
        user.email = user_data.email

    if user_data.password is not None:
        if getattr(user, "auth_source", "local") == "ldap":
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Cannot set password for LDAP users",
            )
        user.password_hash = get_password_hash(user_data.password)
        # As a reset: sessions and pending tokens of the old password end (an
        # admin setting a password is often taking an account back).
        await drop_pending_tokens(db, user.username)
        user.password_changed_at = utcnow_naive()
        close_sockets = True

    if user_data.role is not None:
        if user_data.role not in ["admin", "user"]:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Role must be 'admin' or 'user'",
            )
        user.role = user_data.role

    if user_data.is_active is not None:
        if user.is_active and not user_data.is_active:
            # Deactivated: their printers go at the scope refresh below, but
            # events of no printer (inventory, SpoolBuddy tags, announcements)
            # would still reach their open dashboards, and their tokens would
            # open new ones. The same for their API keys' (a kiosk), which
            # REST already refuses for a deactivated owner.
            from backend.app.api.routes.api_keys import drop_api_key_tokens

            await drop_pending_tokens(db, user.username)
            deactivated_keys = list(
                (await db.execute(select(APIKey.id).where(APIKey.user_id == user_id))).scalars().all()
            )
            await drop_api_key_tokens(db, deactivated_keys)
            close_sockets = True
        user.is_active = user_data.is_active

    # Handle group assignments
    if user_data.group_ids is not None:
        groups_result = await db.execute(select(Group).where(Group.id.in_(user_data.group_ids)))
        groups = groups_result.scalars().all()
        if len(groups) != len(user_data.group_ids):
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="One or more group IDs are invalid",
            )
        user.groups = list(groups)

    await ensure_user_finance_defaults(db, user)

    await db.commit()
    if deactivated_keys:
        await ws_manager.close_api_keys(deactivated_keys)
    if close_sockets:
        # Open dashboards are not signed in again. After a rename they keep the
        # name they connected with and look their printers up by it on every
        # scope refresh (the next user given it); after a password set they are
        # whoever held the account before.
        await ws_manager.close_user(user_id)
    if user_data.group_ids is not None or user_data.role is not None or user_data.is_active is not None:
        # Groups, admin role and deactivation all change which printers the
        # user's open dashboards may hear about (#1727)
        await ws_manager.refresh_printer_scopes()
    result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
    user = result.scalar_one()

    return _user_to_response(user)


@router.get("/{user_id}/items-count")
async def get_user_items_count(
    user_id: int,
    _: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_READ),
    db: AsyncSession = Depends(get_db),
):
    """Get count of items created by this user. Read-only — gated on
    ``USERS_READ`` only."""
    # Verify user exists
    result = await db.execute(select(User).where(User.id == user_id))
    if not result.scalar_one_or_none():
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    # Count archives
    archives_result = await db.execute(select(func.count(PrintArchive.id)).where(PrintArchive.created_by_id == user_id))
    archives_count = archives_result.scalar() or 0

    # Count queue items
    queue_result = await db.execute(
        select(func.count(PrintQueueItem.id)).where(PrintQueueItem.created_by_id == user_id)
    )
    queue_items_count = queue_result.scalar() or 0

    # Count library files
    library_result = await db.execute(
        select(func.count(LibraryFile.id)).where(
            LibraryFile.created_by_id == user_id,
            LibraryFile.deleted_at.is_(None),
        )
    )
    library_files_count = library_result.scalar() or 0

    return {
        "archives": archives_count,
        "queue_items": queue_items_count,
        "library_files": library_files_count,
    }


@router.delete("/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
    user_id: int,
    delete_items: bool = Query(False, description="Delete all items created by this user"),
    _admin: User | None = RequireAdminIfAuthEnabled(),
    current_user: User | None = RequirePermissionIfAuthEnabled(Permission.USERS_DELETE),
    db: AsyncSession = Depends(get_db),
):
    """Delete a user.

    If delete_items=True, all archives, queue items, and library files created by
    this user will also be deleted. Otherwise, these items will become "ownerless"
    (created_by_id set to NULL by the foreign key constraint).
    """
    result = await db.execute(select(User).where(User.id == user_id).options(selectinload(User.groups)))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    # Prevent deleting the last admin
    if user.is_admin:
        # Count admins by role or Administrators group membership
        admin_count_result = await db.execute(select(User).where(User.role == "admin", User.id != user_id))
        other_role_admins = admin_count_result.scalars().all()

        # Also check for users in Administrators group
        admin_group_result = await db.execute(
            select(Group).where(Group.name == "Administrators").options(selectinload(Group.users))
        )
        admin_group = admin_group_result.scalar_one_or_none()
        other_group_admins = [u for u in (admin_group.users if admin_group else []) if u.id != user_id and u.is_active]

        # Combine unique admins
        all_other_admins = {u.id for u in other_role_admins} | {u.id for u in other_group_admins}
        if len(all_other_admins) == 0:
            raise HTTPException(
                status_code=status.HTTP_400_BAD_REQUEST,
                detail="Cannot delete the last admin user",
            )

    # Prevent deleting yourself (only if auth is enabled and we have a current user)
    if current_user and user.id == current_user.id:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Cannot delete your own account",
        )

    # Photo directories of the library rows about to go, resolved while the
    # rows still exist to say which ids they belong to. Removed after the
    # commit, so a failed delete leaves the pictures alone (#3077).
    doomed_library_file_ids: list[int] = []
    if delete_items:
        doomed_library_file_ids = list(
            (await db.execute(select(LibraryFile.id).where(LibraryFile.created_by_id == user_id))).scalars().all()
        )
        # Delete all items created by this user. The print log stays, without
        # the archives it pointed at; archives of other users printed from
        # these library files lose that link (both SET NULL on PostgreSQL).
        from backend.app.services.library_trash import delete_dependent_variants, release_queue_references

        await delete_dependent_variants(db, doomed_library_file_ids)
        await release_queue_references(db, doomed_library_file_ids)
        # A print still running keeps what it needs to be billed when it
        # ends: its queue row and the user's archive it prints are detached
        # (no user), as without delete_items, rather than deleted. Deleted, the
        # completion found no queue row and charged the archive's creator
        # instead (for a reprint someone else, who never printed it), or found
        # no archive and billed nothing, its reservation held for good.
        from sqlalchemy import and_, or_

        from backend.app.models.finance import BudgetReservation

        running_rows = (
            await db.execute(
                select(PrintQueueItem.id, PrintQueueItem.archive_id).where(
                    PrintQueueItem.created_by_id == user_id, PrintQueueItem.status == "printing"
                )
            )
        ).all()
        running_items = [row.id for row in running_rows]
        running_archives = [row.archive_id for row in running_rows if row.archive_id is not None]
        running_archives += (
            (
                await db.execute(
                    select(PrintArchive.id).where(
                        PrintArchive.created_by_id == user_id, PrintArchive.status == "printing"
                    )
                )
            )
            .scalars()
            .all()
        )
        # Also one another user is printing (a reprint of this user's archive,
        # dispatched before the printer's start marks the archive printing):
        # deleted, it took their queue row with it and their print went unbilled.
        own_archives = select(PrintArchive.id).where(PrintArchive.created_by_id == user_id)
        running_archives += (
            (
                await db.execute(
                    select(PrintQueueItem.archive_id).where(
                        PrintQueueItem.status == "printing", PrintQueueItem.archive_id.in_(own_archives)
                    )
                )
            )
            .scalars()
            .all()
        )
        if running_items:
            await db.execute(
                sql_update(PrintQueueItem).where(PrintQueueItem.id.in_(running_items)).values(created_by_id=None)
            )
        if running_archives:
            await db.execute(
                sql_update(PrintArchive)
                .where(PrintArchive.id.in_(running_archives), PrintArchive.created_by_id == user_id)
                .values(created_by_id=None)
            )

        # Budget held for the queue items and archives deleted here: no print
        # completion can settle it any more (it finds no archive), and an
        # active reservation held the cost centre's budget, and kept the cost
        # centre from being deleted, for good. Nothing is charged for those:
        # they go, as the admin chose. (A running print's was detached above.)
        await db.execute(
            sql_update(BudgetReservation)
            .where(
                BudgetReservation.status == "active",
                or_(
                    and_(
                        BudgetReservation.source_type == "print_queue",
                        BudgetReservation.source_id.in_(
                            select(PrintQueueItem.id).where(PrintQueueItem.created_by_id == user_id)
                        ),
                    ),
                    BudgetReservation.print_archive_id.in_(
                        select(PrintArchive.id).where(PrintArchive.created_by_id == user_id)
                    ),
                ),
            )
            .values(status="released", released_at=datetime.now(timezone.utc))
        )
        await db.execute(delete(PrintArchive).where(PrintArchive.created_by_id == user_id))
        await db.execute(delete(PrintQueueItem).where(PrintQueueItem.created_by_id == user_id))
        await db.execute(delete(LibraryFile).where(LibraryFile.created_by_id == user_id))
        await db.execute(delete(PrintBatch).where(PrintBatch.created_by_id == user_id))
    else:
        # Explicitly set created_by_id to NULL for all items (ensures consistent behavior
        # across different database backends, including SQLite without foreign key support).
        # PrintBatch carries the same created_by_id FK with ondelete=SET NULL — admin-deleted
        # users would otherwise leave dangling created_by_id on SQLite (#1295 review nit).
        from sqlalchemy import update

        from backend.app.core.printer_scope import resolve_user_printer_scope

        # An ownerless "any <model>" job may run on any printer of that model;
        # the scheduler held it to this user's printers only while it was
        # theirs (#1727). Stage those jobs instead, so an admin decides where
        # they run rather than the job quietly spreading across the fleet.
        if not (await resolve_user_printer_scope(db, user)).is_unrestricted:
            await db.execute(
                update(PrintQueueItem)
                .where(
                    PrintQueueItem.created_by_id == user_id,
                    PrintQueueItem.printer_id.is_(None),
                    PrintQueueItem.status == "pending",
                )
                .values(manual_start=True)
            )
        await db.execute(update(PrintArchive).where(PrintArchive.created_by_id == user_id).values(created_by_id=None))
        await db.execute(
            update(PrintQueueItem).where(PrintQueueItem.created_by_id == user_id).values(created_by_id=None)
        )
        await db.execute(update(LibraryFile).where(LibraryFile.created_by_id == user_id).values(created_by_id=None))
        await db.execute(update(PrintBatch).where(PrintBatch.created_by_id == user_id).values(created_by_id=None))

    # The print log keeps who started each run as long as the user exists, and
    # outlives them either way (ON DELETE SET NULL, which SQLite doesn't apply).
    await db.execute(sql_update(PrintLogEntry).where(PrintLogEntry.created_by_id == user_id).values(created_by_id=None))

    # Drop API keys owned by this user. The model declares ON DELETE CASCADE
    # so Postgres handles this automatically, but SQLite ships with FK
    # enforcement off (the project's existing pattern — same reason the
    # blocks above set created_by_id = NULL by hand). Without an explicit
    # DELETE here, deleting a user on SQLite would leave their API keys
    # with a dangling user_id and ``_user_from_api_key`` would return None,
    # silently degrading the keys to anonymous (and locking them out of
    # /cloud/* — but the rest of the API would still accept them, which is
    # exactly the orphan-key state the CASCADE was meant to prevent).
    from backend.app.api.routes.api_keys import drop_api_key_tokens

    key_ids = (await db.execute(select(APIKey.id).where(APIKey.user_id == user_id))).scalars().all()
    await drop_api_key_tokens(db, key_ids)
    await db.execute(delete(APIKey).where(APIKey.user_id == user_id))

    # Drop OIDC links, MFA state, and long-lived camera-stream tokens
    # owned by this user. Same SQLite/FK pattern as APIKey above. Without
    # these, deleting a user on SQLite leaves:
    #   - UserOIDCLink: the OIDC callback finds the orphan link, fails to
    #     resolve the (now missing) user, and falls through to
    #     "account_inactive" instead of triggering auto_create (#1285).
    #   - UserTOTP: MFA secrets persist in the DB after the owning user.
    #   - UserOTPCode: pending email OTP codes linger.
    #   - LongLivedToken: per-user camera-stream tokens whose secret_hash
    #     is still valid — verify() would happily match them by lookup
    #     prefix even though the user is gone.
    await db.execute(delete(UserOIDCLink).where(UserOIDCLink.user_id == user_id))
    await db.execute(delete(UserTOTP).where(UserTOTP.user_id == user_id))
    await db.execute(delete(UserOTPCode).where(UserOTPCode.user_id == user_id))
    await db.execute(delete(LongLivedToken).where(LongLivedToken.user_id == user_id))

    # Connected apps, same SQLite/FK pattern. A leftover consent row would
    # skip the consent screen for whoever is next given this user id.
    from sqlalchemy import update as _update

    await db.execute(delete(ConnectedAppGrant).where(ConnectedAppGrant.user_id == user_id))
    await drop_pending_tokens(db, user.username)
    await db.execute(_update(ConnectedApp).where(ConnectedApp.created_by_id == user_id).values(created_by_id=None))

    # Email 2FA is a settings key named after the id, which no foreign key covers.
    await db.execute(delete(Settings).where(Settings.key == f"user_{user_id}_email_2fa_enabled"))

    await db.delete(user)
    await db.commit()
    # What still runs for them must not be credited to whoever gets the id next
    # (SQLite reuses it): a print, a slice, an open dashboard.
    printer_manager.forget_user(user_id)
    from backend.app.main import forget_print_creator

    forget_print_creator(user_id)
    slice_dispatch.forget_owner(user_id)
    await ws_manager.close_user(user_id)
    await ws_manager.close_api_keys(key_ids)
    from backend.app.api.routes.notifications import forget_app_senders

    forget_app_senders(key_ids, [user_id])
    await ws_manager.refresh_printer_scopes()

    for file_id in doomed_library_file_ids:
        remove_library_photos_dir(file_id)


@router.post("/me/change-password", response_model=dict)
async def change_own_password(
    password_data: ChangePasswordRequest,
    credentials: Annotated[HTTPAuthorizationCredentials | None, Depends(security)] = None,
    current_user: User | None = Depends(get_current_user_optional),
    db: AsyncSession = Depends(get_db),
):
    """Change the current user's password. Requires current password verification."""
    if not current_user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Authentication required to change password",
        )

    # Block password change for LDAP users
    if getattr(current_user, "auth_source", "local") == "ldap":
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Cannot change password for LDAP users — passwords are managed by the LDAP server",
        )

    # Verify current password
    if not current_user.password_hash:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Account has no local password set",
        )

    # Rate-limit failed password-change attempts (H-R5-A)
    from backend.app.api.routes.mfa import MAX_2FA_ATTEMPTS, check_rate_limit, record_failed_attempt

    await check_rate_limit(db, current_user.username, event_type="password_change", max_attempts=MAX_2FA_ATTEMPTS)

    if not verify_password(password_data.current_password, current_user.password_hash):
        await record_failed_attempt(db, current_user.username, event_type="password_change")
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Current password is incorrect",
        )

    # Fetch user from this session to ensure changes are persisted
    result = await db.execute(select(User).where(User.id == current_user.id))
    user = result.scalar_one_or_none()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )

    # Update password
    user.password_hash = get_password_hash(password_data.new_password)
    user.password_changed_at = datetime.now(timezone.utc)  # M-R7-B: invalidate all prior JWTs
    # The WebSocket and camera tokens too: reusable for an hour, they would
    # reconnect a closed dashboard straight away.
    await drop_pending_tokens(db, user.username)
    await db.commit()
    # Open dashboards are never signed in again: one someone else had open on
    # the account would keep receiving its printers. This tab signs in again
    # (the route revokes its JWT anyway).
    await ws_manager.close_user(user.id)

    # L-R6-A: Password verified successfully — reset the failure counter
    from backend.app.api.routes.mfa import clear_failed_attempts

    await clear_failed_attempts(db, user.username, event_type="password_change")

    # Revoke the current session token so the caller must re-authenticate (M-R5-A)
    if credentials is not None:
        try:
            payload = _jwt.decode(credentials.credentials, SECRET_KEY, algorithms=[ALGORITHM])
            jti = payload.get("jti")
            exp = payload.get("exp")
            if jti and exp:
                try:
                    await revoke_jti(jti, datetime.fromtimestamp(exp, tz=timezone.utc), user.username)
                except Exception as exc:
                    # B4: log so operators know revocation is broken; password was
                    # already changed so the token will fail freshness checks anyway.
                    import logging

                    logging.getLogger(__name__).error(
                        "Failed to revoke JTI after password change for user %s: %s", user.username, exc
                    )
        except Exception:
            pass  # Decode failure is harmless — token is already invalidated by password_changed_at

    return {"message": "Password changed successfully"}