team-tryouts/app/routes/users/accounts.py

"""Account administration — the president's view of the user list.

Creating, editing, deleting and viewing accounts. Everything here is
admin-only except view_user, which renders a public profile.
"""

from flask import flash, redirect, render_template, request, url_for
from flask_babel import gettext as _
from flask_login import current_user, login_required
from marshmallow import ValidationError

from app.extensions import db, hash_password
from app.logging_config import log_auth_event
from app.models import (
    Admin,
    CoachAvailability,
    Contract,
    ESPORT_GAMES,
    Evaluation,
    GAME_PLATFORMS,
    Match,
    MatchParticipant,
    OneOnOneRequest,
    OrgTeam,
    PersonalNote,
    Player,
    PlayerDisponibility,
    Team,
    TeamMember,
    TeamNote,
    TeamPlayer,
    Tryout,
    TryoutRegistration,
    USER_TYPES,
    User,
    UserGamertag,
)
from app.routes.users._shared import (
    USER_CLASS_MAP,
    flash_validation_errors,
    form_payload,
    update_user_gamertags,
)
from app.routes.users.blueprint import users_bp
from app.validators import CreateUserSchema, EditUserSchema


@users_bp.route('')
@login_required
def list_users():
    """List all users for management (Admin only)."""
    if not isinstance(current_user, Admin):
        flash(_('Only the president can manage users.'), 'danger')
        return redirect(url_for('main.dashboard'))

    users = User.query.order_by(User.role, User.username).all()
    return render_template('pages/users.html', users=users, roles=USER_TYPES)


@users_bp.route('/<int:user_id>/edit', methods=['GET', 'POST'])
@login_required
def edit_user(user_id):
    """Edit an existing user (Admin only)."""
    if not isinstance(current_user, Admin):
        flash(_('Only the president can edit users.'), 'danger')
        return redirect(url_for('main.dashboard'))

    user = User.query.get_or_404(user_id)

    if request.method == 'POST':
        actor_name, actor_id = current_user.username, current_user.id

        def _rerender():
            return render_template(
                'pages/edit_user.html',
                user=user,
                roles=USER_TYPES,
                esport_games=ESPORT_GAMES,
                game_platforms=GAME_PLATFORMS,
                user_gamertags={
                    gt.game: {'gamertag': gt.gamertag, 'platform': gt.platform}
                    for gt in user.gamertags
                },
            )

        try:
            validated = EditUserSchema().load(form_payload(checkboxes=('is_active_account',)))
        except ValidationError as err:
            flash_validation_errors(err)
            return _rerender()

        full_name = validated['full_name']
        email = validated['email']
        phone = validated.get('phone')
        role = validated['role']
        is_active = validated['is_active_account']
        selected_games = validated.get('games', [])
        discord_username = validated.get('discord_username')
        discord_user_id = validated.get('discord_user_id')
        league_os_profile = validated.get('league_os_profile')

        # Previously absent: the column is unique, so assigning a taken
        # address surfaced as an IntegrityError, i.e. a 500.
        clash = User.query.filter(User.email == email, User.id != user.id).first()
        if clash:
            flash(_('Email already in use by another account.'), 'danger')
            return _rerender()

        role_changed = user.role != role
        previous_role = user.role

        if role_changed:
            # Two ways to lock everyone out of administration, neither of
            # which any interface can undo afterwards.
            if user.id == actor_id:
                flash(
                    _('You cannot change your own role. Ask another president to do it.'), 'danger'
                )
                return _rerender()

            if user.role == 'admin':
                remaining_admins = User.query.filter(
                    User.role == 'admin',
                    User.is_active_account.is_(True),
                    User.id != user.id,
                ).count()
                if remaining_admins == 0:
                    flash(
                        _(
                            'This is the last active president. Promote '
                            'another account before changing this one.'
                        ),
                        'danger',
                    )
                    return _rerender()

        if role_changed:
            # The role column is the polymorphic discriminator, and SQLAlchemy
            # decides an instance's class when it loads it. Assigning to it
            # through the ORM leaves a Player object in the identity map for a
            # row that now says 'coach', so every later isinstance() check —
            # which is how this application does authorisation — answers with
            # the old role. Hence the statement-level UPDATE.
            #
            # The instance then has to be re-read. This used to call
            # db.session.remove(), which throws away the whole session:
            # everything the request still held was detached, current_user
            # included, and the next attribute access on any of them raised
            # DetachedInstanceError. Expunging the one stale instance is
            # enough, and it leaves the transaction open — so the role change
            # and the rest of the edit now commit together instead of the
            # role landing on its own and the remaining fields failing after
            # it (ARCH-008).
            user_pk = user.id
            db.session.execute(
                db.text('UPDATE users SET role = :role WHERE id = :id'),
                {'role': role, 'id': user_pk},
            )
            db.session.expunge(user)
            user = db.session.get(User, user_pk)

        user.full_name = full_name
        user.email = email
        user.phone = phone
        user.is_active_account = is_active
        user.games = ','.join(selected_games) if selected_games else None
        user.discord_username = discord_username or None
        user.discord_user_id = discord_user_id or None
        user.league_os_profile = league_os_profile or None

        update_user_gamertags(user, selected_games)

        # Blank means "keep the current password"; anything else has already
        # been checked against the policy by the schema.
        password = validated.get('password')
        if password:
            user.password_hash = hash_password(password)

        db.session.commit()

        # Logged after the commit, not before: the audit trail should record
        # what happened, and until this point nothing had.
        if role_changed:
            log_auth_event(
                'account.role_changed',
                actor=actor_name,
                actor_id=actor_id,
                target=user.username,
                target_id=user.id,
                previous_role=previous_role,
                new_role=role,
            )
        if password:
            log_auth_event(
                'account.password_reset_by_admin',
                actor=actor_name,
                actor_id=actor_id,
                target=user.username,
                target_id=user.id,
            )
        log_auth_event(
            'account.updated',
            actor=actor_name,
            actor_id=actor_id,
            target=user.username,
            target_id=user.id,
            active=is_active,
        )
        flash(_('User %(username)s updated successfully!', username=user.username), 'success')
        return redirect(url_for('users.list_users'))

    user_gamertags = {
        gt.game: {'gamertag': gt.gamertag, 'platform': gt.platform} for gt in user.gamertags
    }
    return render_template(
        'pages/edit_user.html',
        user=user,
        roles=USER_TYPES,
        esport_games=ESPORT_GAMES,
        game_platforms=GAME_PLATFORMS,
        user_gamertags=user_gamertags,
    )


@users_bp.route('/<int:user_id>/delete', methods=['POST'])
@login_required
def delete_user(user_id):
    """Delete a user (Admin only)."""
    if not isinstance(current_user, Admin):
        flash(_('Only the president can delete users.'), 'danger')
        return redirect(url_for('main.dashboard'))

    if current_user.id == user_id:
        flash(_('You cannot delete your own account.'), 'danger')
        return redirect(url_for('users.list_users'))

    user = User.query.get_or_404(user_id)

    Evaluation.query.filter(
        db.or_(Evaluation.evaluator_id == user_id, Evaluation.player_id == user_id),
    ).delete(synchronize_session=False)
    PlayerDisponibility.query.filter_by(player_id=user_id).delete()
    CoachAvailability.query.filter_by(coach_id=user_id).delete()
    PersonalNote.query.filter(
        db.or_(PersonalNote.player_id == user_id, PersonalNote.coach_id == user_id),
    ).delete(synchronize_session=False)
    TeamNote.query.filter_by(coach_id=user_id).delete()
    OneOnOneRequest.query.filter(
        db.or_(OneOnOneRequest.player_id == user_id, OneOnOneRequest.coach_id == user_id),
    ).delete(synchronize_session=False)
    UserGamertag.query.filter_by(user_id=user_id).delete()
    Contract.query.filter_by(player_id=user_id).delete()
    TryoutRegistration.query.filter_by(player_id=user_id).delete()
    TeamPlayer.query.filter_by(player_id=user_id).delete()
    TeamMember.query.filter_by(player_id=user_id).delete()
    MatchParticipant.query.filter_by(player_id=user_id).delete()
    OrgTeam.query.filter_by(coach_id=user_id).update({'coach_id': None})
    OrgTeam.query.filter_by(manager_id=user_id).update({'manager_id': None})
    Tryout.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
    Match.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
    Team.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
    OrgTeam.query.filter_by(created_by=user_id).update({'created_by': current_user.id})
    Contract.query.filter_by(uploaded_by_id=user_id).update({'uploaded_by_id': current_user.id})

    deleted_username, deleted_role = user.username, user.role
    db.session.delete(user)
    db.session.commit()
    log_auth_event(
        'account.deleted',
        actor=current_user.username,
        actor_id=current_user.id,
        target=deleted_username,
        target_id=user_id,
        role=deleted_role,
    )
    flash(
        _('User %(deleted_username)s has been removed.', deleted_username=deleted_username),
        'success',
    )
    return redirect(url_for('users.list_users'))


@users_bp.route('/create', methods=['GET', 'POST'])
@login_required
def create_user():
    """Create a new user (Admin only). Uses the correct polymorphic subclass."""
    if not isinstance(current_user, Admin):
        flash(_('Only the president can create users.'), 'danger')
        return redirect(url_for('main.dashboard'))

    if request.method == 'POST':
        try:
            validated = CreateUserSchema().load(request.form)
        except ValidationError as err:
            flash_validation_errors(err)
            return render_template('pages/create_user.html', roles=USER_TYPES)

        username = validated['username']
        email = validated['email']
        password = validated['password']
        full_name = validated['full_name']
        phone = validated.get('phone')
        # The schema constrains role with OneOf(USER_TYPES), so the former
        # manual membership check is now redundant.
        role = validated['role']

        if User.query.filter_by(username=username).first():
            flash(_('Username already exists.'), 'danger')
            return render_template('pages/create_user.html', roles=USER_TYPES)

        if User.query.filter_by(email=email).first():
            flash(_('Email already registered.'), 'danger')
            return render_template('pages/create_user.html', roles=USER_TYPES)

        hashed_password = hash_password(password)
        user_cls = USER_CLASS_MAP.get(role, Player)
        user = user_cls(
            username=username,
            password_hash=hashed_password,
            role=role,
            full_name=full_name,
            email=email,
            phone=phone,
        )
        db.session.add(user)
        db.session.commit()
        log_auth_event(
            'account.created_by_admin',
            actor=current_user.username,
            actor_id=current_user.id,
            target=user.username,
            target_id=user.id,
            role=role,
        )
        flash(
            _('User %(full_name)s created as %(role)s!', full_name=full_name, role=role), 'success'
        )
        return redirect(url_for('users.list_users'))

    return render_template('pages/create_user.html', roles=USER_TYPES)


@users_bp.route('/<int:user_id>/view')
@login_required
def view_user(user_id):
    """View a public profile for any user."""
    user = User.query.get_or_404(user_id)
    return render_template('pages/view_user.html', profile_user=user)