Support API-driven avatar changes: login.avatar_id sync + refresh consumer

.github/workflows/test.yml

@@ -12,7 +12,7 @@ on:
     - cron: '0 0 * * *'
 
 env:
-  FAF_DB_VERSION: v138
+  FAF_DB_VERSION: v143
   FLYWAY_VERSION: 7.5.4
 
 jobs:

compose.yaml

@@ -16,7 +16,7 @@ services:
       - "3306:3306"
 
   faf-db-migrations:
-    image: faforever/faf-db-migrations:v138
+    image: faforever/faf-db-migrations:v143
     command: migrate
     environment:
       FLYWAY_URL: jdbc:mysql://faf-db/faf?useSSL=false

server/__init__.py

@@ -117,6 +117,7 @@
 import server.metrics as metrics
 
 from .asyncio_extensions import map_suppress, synchronizedmethod
+from .avatar_change_queue_service import AvatarChangeQueueService
 from .broadcast_service import BroadcastService
 from .client_message_queue_service import ClientMessageQueueService
 from .config import TRACE, config
@@ -144,6 +145,7 @@
 __copyright__ = "Copyright (c) 2011-2015 " + __author__
 
 __all__ = (
+    "AvatarChangeQueueService",
     "BroadcastService",
     "ClientMessageQueueService",
     "ConfigurationService",

server/avatar_change_queue_service.py

@@ -0,0 +1,109 @@
+"""
+Consume "player avatar changed" events from trusted microservices.
+
+# Wire contract
+Publishers post to the `MQ_EXCHANGE_NAME` topic exchange with routing key
+`success.player_avatar.update`. The body is a UTF-8 JSON object:
+
+- `player_id` (int, required): the player whose selected avatar changed.
+- `avatar_id` (int or null, optional): the newly selected avatar id, or
+  null if the player cleared their avatar. The lobby itself ignores
+  this field — it always re-reads the DB so it gets the matching
+  `url`/`tooltip` and applies the ownership check. The field is
+  shipped for the benefit of *other* subscribers that may want to act
+  on the change without an extra DB roundtrip.
+
+On receipt the lobby re-reads the affected player's avatar from the DB
+and marks them dirty so the existing `BroadcastService` emits a
+`player_info` to every connected client on its next tick.
+"""
+
+import json
+import logging
+import socket
+from typing import ClassVar, Optional
+
+from aio_pika.abc import AbstractIncomingMessage, AbstractQueue
+
+from .config import config
+from .core import Service
+from .decorators import with_logger
+from .message_queue_service import MessageQueueService
+from .player_service import PlayerService
+
+
+PLAYER_AVATAR_UPDATE_ROUTING_KEY = "success.player_avatar.update"
+
+
+@with_logger
+class AvatarChangeQueueService(Service):
+    """Consume `success.player_avatar.update` messages and refresh players."""
+
+    _logger: ClassVar[logging.Logger]
+
+    def __init__(
+        self,
+        message_queue_service: MessageQueueService,
+        player_service: PlayerService,
+    ):
+        self.message_queue_service = message_queue_service
+        self.player_service = player_service
+        self._queue: Optional[AbstractQueue] = None
+        self._consumer_tag: Optional[str] = None
+
+    async def initialize(self) -> None:
+        # Per-instance queue: every lobby pod must process every event so
+        # whichever pod is hosting the player can refresh its in-memory
+        # state. Naming follows `<exchange>.<service>.<routing-key>.<host>`,
+        # matching `ClientMessageQueueService`.
+        queue_name = (
+            f"{config.MQ_EXCHANGE_NAME}.lobby.player_avatar.update"
+            f".{socket.gethostname()}"
+        )
+        result = await self.message_queue_service.declare_queue_and_consume(
+            exchange_name=config.MQ_EXCHANGE_NAME,
+            routing_key=PLAYER_AVATAR_UPDATE_ROUTING_KEY,
+            callback=self._on_message,
+            queue_name=queue_name,
+        )
+        if result is not None:
+            self._queue, self._consumer_tag = result
+
+    async def shutdown(self) -> None:
+        if self._queue is not None and self._consumer_tag is not None:
+            await self._queue.cancel(self._consumer_tag)
+        self._queue = None
+        self._consumer_tag = None
+
+    async def _on_message(self, message: AbstractIncomingMessage) -> None:
+        async with message.process(requeue=False):
+            try:
+                payload = json.loads(message.body)
+            except (ValueError, UnicodeDecodeError):
+                self._logger.warning(
+                    "Dropping avatar-update message with non-JSON body"
+                )
+                return
+
+            if not isinstance(payload, dict):
+                self._logger.warning(
+                    "Dropping avatar-update message: payload is not a JSON object"
+                )
+                return
+
+            raw_player_id = payload.get("player_id")
+            try:
+                player_id = int(raw_player_id)
+            except (TypeError, ValueError):
+                self._logger.warning(
+                    "Dropping avatar-update message: invalid player_id %r",
+                    raw_player_id,
+                )
+                return
+
+            refreshed = await self.player_service.refresh_player_avatar(player_id)
+            if not refreshed:
+                self._logger.debug(
+                    "avatar-update for player %s ignored: not connected here",
+                    player_id,
+                )

server/db/models.py

@@ -184,7 +184,8 @@
     Column("create_time",   TIMESTAMP,  nullable=False),
     Column("update_time",   TIMESTAMP,  nullable=False),
     Column("user_agent",    String),
-    Column("last_login",    TIMESTAMP)
+    Column("last_login",    TIMESTAMP),
+    Column("avatar_id",     Integer,    ForeignKey("avatars_list.id")),
 )
 
 leaderboard = Table(

server/lobbyconnection.py

@@ -957,6 +957,7 @@ async def command_avatar(self, message):
                 )
                 self.player.avatar = None
 
+                new_avatar_id = row.id if avatar_url is not None else None
                 if avatar_url is not None:
                     await conn.execute(
                         avatars.update().where(
@@ -972,6 +973,15 @@ async def command_avatar(self, message):
                         "url": avatar_url,
                         "tooltip": row.tooltip
                     }
+                # Mirror the selection to login.avatar_id so reads via the new
+                # authoritative column stay consistent with the legacy flag.
+                await conn.execute(
+                    t_login.update().where(
+                        t_login.c.id == self.player.id
+                    ).values(
+                        avatar_id=new_avatar_id
+                    )
+                )
                 self.player_service.mark_dirty(self.player)
         else:
             raise KeyError("invalid action")

server/player_service.py

@@ -8,7 +8,7 @@
 from typing import TYPE_CHECKING, ClassVar, Optional, ValuesView
 
 import aiocron
-from sqlalchemy import and_, select
+from sqlalchemy import and_, or_, select
 
 import server.metrics as metrics
 from server.config import config
@@ -90,6 +90,12 @@
             )
             player.user_groups = {row.technical_name for row in result}
 
+            # Avatar lookup: `login.avatar_id` is the new authoritative FK,
+            # with a fallback to the legacy `avatars.selected = 1` row when
+            # `avatar_id` is null. In both cases we route through the
+            # `avatars` grant table so ownership is enforced — a stale
+            # `login.avatar_id` pointing at a no-longer-granted avatar
+            # yields no row here.
             sql = select(
                 avatars_list.c.url,
                 avatars_list.c.tooltip,
@@ -102,7 +108,13 @@
                     avatars,
                     onclause=and_(
                         avatars.c.idUser == login.c.id,
-                        avatars.c.selected == 1
+                        or_(
+                            avatars.c.idAvatar == login.c.avatar_id,
+                            and_(
+                                login.c.avatar_id.is_(None),
+                                avatars.c.selected == 1
+                            )
+                        )
                     )
                 )
                 .outerjoin(avatars_list)
@@ -129,6 +141,57 @@
 
             await self._fetch_player_ratings(player, conn)
 
+    async def _fetch_player_avatar(self, player: Player, conn) -> None:
+        # Same ownership-checked join as `fetch_player_data`'s avatar block:
+        # trust `login.avatar_id` only when a matching grant row exists in
+        # `avatars`, with a fallback to the legacy `selected = 1` flag.
+        sql = select(
+            avatars_list.c.url,
+            avatars_list.c.tooltip,
+        ).select_from(
+            login
+            .outerjoin(
+                avatars,
+                onclause=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
+                        )
+                    )
+                )
+            )
+            .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
+
+        row_mapping = row._mapping
+        url = row_mapping.get(avatars_list.c.url)
+        tooltip = row_mapping.get(avatars_list.c.tooltip)
+        player.avatar = (
+            {"url": url, "tooltip": tooltip} if url and tooltip else None
+        )
+
+    async def refresh_player_avatar(self, player_id: int) -> bool:
+        """Re-read avatar from DB for one player and mark them dirty so
+        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:
+            await self._fetch_player_avatar(player, conn)
+        self.mark_dirty(player)
+        return True
+
     async def _fetch_player_ratings(self, player: Player, conn):
         sql = select(
             leaderboard_rating.c.mean,