Module server.player_service

Manages connected and authenticated players

Classes

class PlayerService (database: FAFDatabase)

Expand source code

@with_logger
class PlayerService(Service):
    _logger: ClassVar[logging.Logger]

    def __init__(self, database: FAFDatabase):
        self._db = database
        self._players: dict[int, Player] = {}

        # Static-ish data fields.
        self.uniqueid_exempt: frozenset[int] = frozenset()
        self._dirty_players: set[Player] = set()

    async def initialize(self) -> None:
        await self.update_data()
        self._update_cron = aiocron.crontab(
            "*/10 * * * *", func=self.update_data
        )

    def __len__(self) -> int:
        return len(self._players)

    def __iter__(self) -> Iterator[Player]:
        return self._players.values().__iter__()

    def __getitem__(self, player_id: int) -> Optional[Player]:
        return self._players.get(player_id)

    def __setitem__(self, player_id: int, player: Player) -> None:
        self._players[player_id] = player
        metrics.players_online.set(len(self._players))

    @property
    def all_players(self) -> ValuesView[Player]:
        return self._players.values()

    def mark_dirty(self, player: Player) -> None:
        self._dirty_players.add(player)

    def pop_dirty_players(self) -> set[Player]:
        dirty_players = self._dirty_players
        self._dirty_players = set()

        return dirty_players

    @staticmethod
    def _avatar_grant_join_onclause():
        # ON clause for joining `avatars` against `login` to pick a
        # player's currently worn avatar with ownership enforced.
        # Prefer the new authoritative `login.avatar_id` column; fall
        # back to the legacy `avatars.selected = 1` row only when
        # `avatar_id` is null. Either way the row must be a real grant
        # in `avatars`, so revoked grants resolve to no avatar.
        return and_(
            avatars.c.idUser == login.c.id,
            or_(
                avatars.c.idAvatar == login.c.avatar_id,
                and_(
                    login.c.avatar_id.is_(None),
                    avatars.c.selected == 1
                )
            )
        )

    async def fetch_player_data(self, player: Player) -> None:
        async with self._db.acquire() as conn:
            result = await conn.execute(
                select(user_group.c.technical_name)
                .select_from(user_group_assignment.join(user_group))
                .where(user_group_assignment.c.user_id == player.id)
            )
            player.user_groups = {row.technical_name for row in result}

            sql = select(
                avatars_list.c.url,
                avatars_list.c.tooltip,
                clan.c.tag
            ).select_from(
                login
                .outerjoin(clan_membership)
                .outerjoin(clan)
                .outerjoin(
                    avatars,
                    onclause=self._avatar_grant_join_onclause()
                )
                .outerjoin(avatars_list)
            ).where(login.c.id == player.id)  # yapf: disable

            result = await conn.execute(sql)
            row = result.fetchone()
            if not row:
                self._logger.warning(
                    "Did not find data for player with id %i",
                    player.id
                )
                return

            row_mapping = row._mapping
            player.clan = row_mapping.get(clan.c.tag)

            url, tooltip = (
                row_mapping.get(avatars_list.c.url),
                row_mapping.get(avatars_list.c.tooltip)
            )
            if url and tooltip:
                player.avatar = {"url": url, "tooltip": tooltip}

            await self._fetch_player_ratings(player, conn)

    async def _fetch_player_avatar(
        self, player: Player, conn
    ) -> Optional[int]:
        """Refresh `player.avatar` from DB; return the avatar id, if any."""
        sql = select(
            avatars_list.c.id,
            avatars_list.c.url,
            avatars_list.c.tooltip,
        ).select_from(
            login
            .outerjoin(
                avatars,
                onclause=self._avatar_grant_join_onclause()
            )
            .outerjoin(avatars_list)
        ).where(login.c.id == player.id)

        result = await conn.execute(sql)
        row = result.fetchone()
        if row is None:
            player.avatar = None
            return None

        row_mapping = row._mapping
        avatar_id = row_mapping.get(avatars_list.c.id)
        url = row_mapping.get(avatars_list.c.url)
        tooltip = row_mapping.get(avatars_list.c.tooltip)
        if url and tooltip:
            player.avatar = {"url": url, "tooltip": tooltip}
            return avatar_id
        player.avatar = None
        return None

    async def _sync_legacy_avatar_selection(self, player_id: int, conn) -> None:
        """
        Reconcile the deprecated `avatars.selected` flag with the authoritative
        `login.avatar_id`: mark only the granted row matching `avatar_id` as
        selected (all of them deselected when `avatar_id` is null). This keeps the
        legacy fallback consistent so an explicit clear is not resurrected.
        """
        authoritative_avatar_id = (
            select(login.c.avatar_id)
            .where(login.c.id == player_id)
            .scalar_subquery()
        )
        await conn.execute(
            avatars.update()
            .where(avatars.c.idUser == player_id)
            .values(
                selected=func.coalesce(
                    avatars.c.idAvatar == authoritative_avatar_id, False
                )
            )
        )

    async def refresh_player_avatar(self, player_id: int) -> bool:
        """
        Re-read avatar for one player and mark them dirty.

        `BroadcastService` emits a `player_info` on the next tick. Returns
        True if the player is connected to this instance, False otherwise.
        """
        player = self._players.get(player_id)
        if player is None:
            return False
        async with self._db.acquire() as conn:
            # An avatar-update event means the API (the authoritative writer of
            # `login.avatar_id`) changed this player's selection, so reconcile the
            # deprecated `avatars.selected` flag before reading. Otherwise a cleared
            # avatar (`avatar_id` set to null) would be undone by the legacy
            # fallback in `_avatar_grant_join_onclause`.
            await self._sync_legacy_avatar_selection(player_id, conn)
            avatar_id = await self._fetch_player_avatar(player, conn)
        avatar_tooltip = player.avatar["tooltip"] if player.avatar else None
        self._logger.info(
            "Player %s avatar refreshed from RabbitMQ event: "
            "avatar_id=%s tooltip=%s",
            player_id, avatar_id, avatar_tooltip
        )
        self.mark_dirty(player)
        return True

    async def _fetch_player_ratings(self, player: Player, conn):
        sql = select(
            leaderboard_rating.c.mean,
            leaderboard_rating.c.deviation,
            leaderboard_rating.c.total_games,
            leaderboard.c.technical_name,
        ).select_from(
            leaderboard.join(leaderboard_rating)
        ).where(
            leaderboard_rating.c.login_id == player.id
        )
        result = await conn.execute(sql)

        retrieved_ratings = {
            row.technical_name: (
                (row.mean, row.deviation),
                row.total_games
            )
            for row in result
        }
        for rating_type, (rating, total_games) in retrieved_ratings.items():
            player.ratings[rating_type] = rating
            player.game_count[rating_type] = total_games

    def remove_player(self, player: Player) -> None:
        if player.id in self._players:
            # This signals that the player is now disconnected
            del player.lobby_connection
            del self._players[player.id]
            metrics.players_online.set(len(self._players))
            self.mark_dirty(player)

    async def has_permission_role(self, player: Player, role_name: str) -> bool:
        async with self._db.acquire() as conn:
            result = await conn.execute(
                select(group_permission.c.id)
                .select_from(
                    user_group_assignment
                    .join(group_permission_assignment, onclause=(
                        user_group_assignment.c.group_id ==
                        group_permission_assignment.c.group_id
                    ))
                    .join(group_permission)
                )
                .where(
                    and_(
                        user_group_assignment.c.user_id == player.id,
                        group_permission.c.technical_name == role_name
                    )
                )
            )
            row = result.fetchone()
            return row is not None

    def is_uniqueid_exempt(self, user_id: int) -> bool:
        return user_id in self.uniqueid_exempt

    def get_player(self, player_id: int) -> Optional[Player]:
        return self._players.get(player_id)

    def signal_player_rating_change(
        self, player_id: int, rating_type: str, new_rating: Rating
    ) -> None:
        player = self.get_player(player_id)
        if player is None:
            self._logger.debug(
                "Received rating change for player with id %i not in PlayerService.",
                player_id
            )
            return

        self._logger.debug(
            "Received rating change for player %s.", player
        )
        player.ratings[rating_type] = new_rating
        player.game_count[rating_type] += 1
        self.mark_dirty(player)

    async def update_data(self) -> None:
        """
        Update rarely-changing data, such as the admin list and the list of users exempt from the
        uniqueid check.
        """
        async with self._db.acquire() as conn:
            # UniqueID-exempt users.
            result = await conn.execute(
                "SELECT `user_id` FROM uniqueid_exempt"
            )
            self.uniqueid_exempt = frozenset(x[0] for x in result)

    async def kick_idle_players(self) -> None:
        for fut in asyncio.as_completed([
            player.lobby_connection.abort("Graceful shutdown.")
            for player in self.all_players
            if player.state == PlayerState.IDLE
            if player.lobby_connection is not None
        ]):
            try:
                await fut
            except Exception:
                self._logger.debug(
                    "Error while aborting connection",
                    exc_info=True
                )

    def on_connection_lost(self, conn: "LobbyConnection") -> None:
        if not conn.player:
            return

        self.remove_player(conn.player)

        self._logger.debug(
            "Removed player %d, %s, %d",
            conn.player.id,
            conn.player.login,
            conn.session
        )

    async def graceful_shutdown(self) -> None:
        if config.SHUTDOWN_KICK_IDLE_PLAYERS:
            self._kick_idle_task = at_interval(1, self.kick_idle_players)

All services should inherit from this class.

Services are singleton objects which manage some server task.

Ancestors

• Service

Instance variables

prop all_players : ValuesView[Player]

Expand source code

@property
def all_players(self) -> ValuesView[Player]:
    return self._players.values()

Methods

async def fetch_player_data(self,
player: Player) ‑> None

Expand source code

async def fetch_player_data(self, player: Player) -> None:
    async with self._db.acquire() as conn:
        result = await conn.execute(
            select(user_group.c.technical_name)
            .select_from(user_group_assignment.join(user_group))
            .where(user_group_assignment.c.user_id == player.id)
        )
        player.user_groups = {row.technical_name for row in result}

        sql = select(
            avatars_list.c.url,
            avatars_list.c.tooltip,
            clan.c.tag
        ).select_from(
            login
            .outerjoin(clan_membership)
            .outerjoin(clan)
            .outerjoin(
                avatars,
                onclause=self._avatar_grant_join_onclause()
            )
            .outerjoin(avatars_list)
        ).where(login.c.id == player.id)  # yapf: disable

        result = await conn.execute(sql)
        row = result.fetchone()
        if not row:
            self._logger.warning(
                "Did not find data for player with id %i",
                player.id
            )
            return

        row_mapping = row._mapping
        player.clan = row_mapping.get(clan.c.tag)

        url, tooltip = (
            row_mapping.get(avatars_list.c.url),
            row_mapping.get(avatars_list.c.tooltip)
        )
        if url and tooltip:
            player.avatar = {"url": url, "tooltip": tooltip}

        await self._fetch_player_ratings(player, conn)

def get_player(self, player_id: int) ‑> Player | None

Expand source code

def get_player(self, player_id: int) -> Optional[Player]:
    return self._players.get(player_id)

async def has_permission_role(self,
player: Player,
role_name: str) ‑> bool

Expand source code

async def has_permission_role(self, player: Player, role_name: str) -> bool:
    async with self._db.acquire() as conn:
        result = await conn.execute(
            select(group_permission.c.id)
            .select_from(
                user_group_assignment
                .join(group_permission_assignment, onclause=(
                    user_group_assignment.c.group_id ==
                    group_permission_assignment.c.group_id
                ))
                .join(group_permission)
            )
            .where(
                and_(
                    user_group_assignment.c.user_id == player.id,
                    group_permission.c.technical_name == role_name
                )
            )
        )
        row = result.fetchone()
        return row is not None

def is_uniqueid_exempt(self, user_id: int) ‑> bool

Expand source code

def is_uniqueid_exempt(self, user_id: int) -> bool:
    return user_id in self.uniqueid_exempt

async def kick_idle_players(self) ‑> None

Expand source code

async def kick_idle_players(self) -> None:
    for fut in asyncio.as_completed([
        player.lobby_connection.abort("Graceful shutdown.")
        for player in self.all_players
        if player.state == PlayerState.IDLE
        if player.lobby_connection is not None
    ]):
        try:
            await fut
        except Exception:
            self._logger.debug(
                "Error while aborting connection",
                exc_info=True
            )

def mark_dirty(self,
player: Player) ‑> None

Expand source code

def mark_dirty(self, player: Player) -> None:
    self._dirty_players.add(player)

def pop_dirty_players(self) ‑> set[Player]

Expand source code

def pop_dirty_players(self) -> set[Player]:
    dirty_players = self._dirty_players
    self._dirty_players = set()

    return dirty_players

async def refresh_player_avatar(self, player_id: int) ‑> bool

Expand source code

async def refresh_player_avatar(self, player_id: int) -> bool:
    """
    Re-read avatar for one player and mark them dirty.

    `BroadcastService` emits a `player_info` on the next tick. Returns
    True if the player is connected to this instance, False otherwise.
    """
    player = self._players.get(player_id)
    if player is None:
        return False
    async with self._db.acquire() as conn:
        # An avatar-update event means the API (the authoritative writer of
        # `login.avatar_id`) changed this player's selection, so reconcile the
        # deprecated `avatars.selected` flag before reading. Otherwise a cleared
        # avatar (`avatar_id` set to null) would be undone by the legacy
        # fallback in `_avatar_grant_join_onclause`.
        await self._sync_legacy_avatar_selection(player_id, conn)
        avatar_id = await self._fetch_player_avatar(player, conn)
    avatar_tooltip = player.avatar["tooltip"] if player.avatar else None
    self._logger.info(
        "Player %s avatar refreshed from RabbitMQ event: "
        "avatar_id=%s tooltip=%s",
        player_id, avatar_id, avatar_tooltip
    )
    self.mark_dirty(player)
    return True

Re-read avatar for one player and mark them dirty.

BroadcastService emits a player_info on the next tick. Returns True if the player is connected to this instance, False otherwise.

def remove_player(self,
player: Player) ‑> None

Expand source code

def remove_player(self, player: Player) -> None:
    if player.id in self._players:
        # This signals that the player is now disconnected
        del player.lobby_connection
        del self._players[player.id]
        metrics.players_online.set(len(self._players))
        self.mark_dirty(player)

def signal_player_rating_change(self,
player_id: int,
rating_type: str,
new_rating: Rating) ‑> None

Expand source code

def signal_player_rating_change(
    self, player_id: int, rating_type: str, new_rating: Rating
) -> None:
    player = self.get_player(player_id)
    if player is None:
        self._logger.debug(
            "Received rating change for player with id %i not in PlayerService.",
            player_id
        )
        return

    self._logger.debug(
        "Received rating change for player %s.", player
    )
    player.ratings[rating_type] = new_rating
    player.game_count[rating_type] += 1
    self.mark_dirty(player)

async def update_data(self) ‑> None

Expand source code

async def update_data(self) -> None:
    """
    Update rarely-changing data, such as the admin list and the list of users exempt from the
    uniqueid check.
    """
    async with self._db.acquire() as conn:
        # UniqueID-exempt users.
        result = await conn.execute(
            "SELECT `user_id` FROM uniqueid_exempt"
        )
        self.uniqueid_exempt = frozenset(x[0] for x in result)

Update rarely-changing data, such as the admin list and the list of users exempt from the uniqueid check.

Inherited members

• Service:
  • graceful_shutdown
  • initialize
  • on_connection_lost
  • shutdown