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
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, TokenType
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_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.utils.library_paths import remove_library_photos_dir

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]


@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."""
    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",
            )
        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)

    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:
        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 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
        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))

    # 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).
    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, and a
    # code issued in the last minute is keyed by username, not id.
    from sqlalchemy import update as _update

    await db.execute(delete(ConnectedAppGrant).where(ConnectedAppGrant.user_id == user_id))
    await db.execute(
        delete(AuthEphemeralToken).where(
            AuthEphemeralToken.token_type == TokenType.CONNECT_CODE,
            func.lower(AuthEphemeralToken.username) == user.username.lower(),
        )
    )
    await db.execute(_update(ConnectedApp).where(ConnectedApp.created_by_id == user_id).values(created_by_id=None))

    await db.delete(user)
    await db.commit()
    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
    await db.commit()

    # 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"}