Animate individual PvB and PvP moves in lobby live viewer

webapp-dao/src/main/kotlin/io/elephantchess/db/services/PlayerVsBotGameDaoService.kt

@@ -228,6 +228,7 @@ class PlayerVsBotGameDaoService(private val dslContext: DSLContext) {
                     BOT_GAME.ID,
                     BOT_GAME.GAME_STATUS,
                     BOT_GAME.CURRENT_FEN,
+                    BOT_GAME.CURRENT_HALF_MOVE_INDEX,
                     BOT_GAME.LAST_UPDATED
                 )
                 .from(BOT_GAME)

webapp-dao/src/main/kotlin/io/elephantchess/db/services/PlayerVsPlayerGameDaoService.kt

@@ -407,6 +407,7 @@ class PlayerVsPlayerGameDaoService(private val dslContext: DSLContext) {
                     GAME.ID,
                     GAME.GAME_STATUS,
                     GAME.CURRENT_FEN,
+                    GAME.CURRENT_HALF_MOVE_INDEX,
                     GAME.LAST_UPDATED
                 )
                 .from(GAME)
@@ -415,6 +416,34 @@ class PlayerVsPlayerGameDaoService(private val dslContext: DSLContext) {
         }
     }
 
+    /**
+     * @param tuples list of gameId and half move index to request from
+     * @return list of moves for each gameId, position and uci
+     */
+    suspend fun fetchNewMovesForGamesAndIndexes(tuples: List<Pair<String, Int>>): List<GameMove> {
+        if (tuples.isEmpty()) {
+            return emptyList()
+        }
+
+        fun conditionForPair(gameId: String, moveIndex: Int) =
+            GAME_MOVE.GAME_ID.eq(gameId).and(GAME_MOVE.POSITION.ge(moveIndex))
+
+        val conditionForPairs =
+            tuples
+                .map { (gameId, moveIndex) -> conditionForPair(gameId, moveIndex) }
+                .reduce { acc, condition -> acc.or(condition) }
+
+        return dslContext
+            .select(
+                GAME_MOVE.GAME_ID,
+                GAME_MOVE.POSITION,
+                GAME_MOVE.UCI
+            )
+            .from(GAME_MOVE)
+            .where(conditionForPairs)
+            .awaitMappedRecords()
+    }
+
     suspend fun fetchRatingForUser(userId: String, timeControlCategory: TimeControlCategory, variant: Variant): Int? {
         return fetchRatingForUser(dslContext, userId, timeControlCategory, variant)
     }

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/dto/lobby/LatestGamesUpdateRequest.kt

@@ -2,4 +2,7 @@ package io.elephantchess.servicelayer.dto.lobby
 
 import io.elephantchess.model.GameId
 
-data class LatestGamesUpdateRequest(val gameIds: List<GameId>)
+data class LatestGamesUpdateRequest(
+    val gameIds: List<GameId>,
+    val moveIndexes: Map<String, Int> = emptyMap()
+)

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/dto/lobby/LatestGamesUpdateResponse.kt

@@ -9,7 +9,9 @@ data class LatestGamesUpdateResponse(val entries: List<Entry>) {
         val gameId: GameId,
         val status: GameEventType,
         val fen: String,
-        val lastUpdated : Long
+        val lastUpdated: Long,
+        val moveIndex: Int? = null,
+        val newMoves: List<String> = emptyList()
     )
 
 }

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/dto/ws/LiveGamesSubscription.kt

@@ -0,0 +1,13 @@
+package io.elephantchess.servicelayer.dto.ws
+
+import io.elephantchess.model.GameId
+
+/**
+ * Sent by a lobby client over the live-games WebSocket to declare which games it
+ * is currently displaying and wants to receive updates for. The move index per
+ * game is tracked server-side (in [io.elephantchess.servicelayer.services.ws.LiveGamesWebSocketSession]),
+ * so the client only needs to send the game ids.
+ */
+data class LiveGamesSubscription(
+    val gameIds: List<GameId> = emptyList()
+)

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/services/GameDataService.kt

@@ -736,24 +736,56 @@ class GameDataService(
     suspend fun fetchLatestGamesUpdate(request: LatestGamesUpdateRequest): LatestGamesUpdateResponse {
         val idsByType = request
             .gameIds
-            .groupBy({ gameID -> gameID.type }, { gameId -> gameId.id })
+            .groupBy({ gameId -> gameId.type }, { gameId -> gameId.id })
 
         val pvpGames = pvpGameDaoService.fetchCurrentStatusAndFen(idsByType[PVP].orEmpty())
         val pvbGames = pvbGameDaoService.fetchCurrentStatusAndFen(idsByType[PVB].orEmpty())
 
+        val pvpNewMoves = if (request.moveIndexes.isNotEmpty()) {
+            val moveTuples = pvpGames
+                .filter { game -> request.moveIndexes.containsKey(game.id) }
+                .map { game -> game.id to request.moveIndexes[game.id]!! }
+            pvpGameDaoService.fetchNewMovesForGamesAndIndexes(moveTuples)
+                .groupBy { it.gameId }
+        } else {
+            emptyMap()
+        }
+
+        val pvbNewMoves = if (request.moveIndexes.isNotEmpty()) {
+            val moveTuples = pvbGames
+                .filter { game -> request.moveIndexes.containsKey(game.id) }
+                .map { game -> game.id to request.moveIndexes[game.id]!! }
+            pvbGameDaoService.fetchNewMovesForGamesAndIndexes(moveTuples)
+                .groupBy { it.botGameId }
+        } else {
+            emptyMap()
+        }
+
         val entries = pvpGames.map { game ->
+            val newMoves = pvpNewMoves[game.id]
+                ?.sortedBy { it.position }
+                ?.map { it.uci }
+                ?: emptyList()
             LatestGamesUpdateResponse.Entry(
                 gameId = GameId(PVP, game.id),
                 status = game.gameStatus,
                 fen = game.currentFen,
-                lastUpdated = game.lastUpdated.toEpochMilliseconds()
+                lastUpdated = game.lastUpdated.toEpochMilliseconds(),
+                moveIndex = game.currentHalfMoveIndex,
+                newMoves = newMoves
             )
         } + pvbGames.map { game ->
+            val newMoves = pvbNewMoves[game.id]
+                ?.sortedBy { it.position }
+                ?.map { it.uci }
+                ?: emptyList()
             LatestGamesUpdateResponse.Entry(
                 gameId = GameId(PVB, game.id),
                 status = game.gameStatus,
                 fen = game.currentFen,
-                lastUpdated = game.lastUpdated.toEpochMilliseconds()
+                lastUpdated = game.lastUpdated.toEpochMilliseconds(),
+                moveIndex = game.currentHalfMoveIndex,
+                newMoves = newMoves
             )
         }
 

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/services/LobbyService.kt

@@ -1,13 +1,40 @@
 package io.elephantchess.servicelayer.services
 
 import io.elephantchess.db.services.UpcomingEventDaoService
+import io.elephantchess.model.GameId
 import io.elephantchess.servicelayer.dto.lobby.GetUpcomingEventsResponse
 import io.elephantchess.servicelayer.dto.lobby.GetUpcomingEventsResponse.UpcomingEvent
+import io.elephantchess.servicelayer.dto.lobby.LatestGamesUpdateRequest
+import io.elephantchess.servicelayer.dto.lobby.LatestGamesUpdateResponse
+import io.elephantchess.servicelayer.services.ws.LiveGamesBatchUpdate
+import io.elephantchess.servicelayer.services.ws.LiveGamesWebSocketSession
+import io.elephantchess.servicelayer.utils.ops.launchAtFixedRate
+import io.github.oshai.kotlinlogging.KLogger
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.ChannelResult
+import kotlin.time.Duration.Companion.seconds
 
 class LobbyService(
-    private val upcomingEventDaoService: UpcomingEventDaoService
+    private val upcomingEventDaoService: UpcomingEventDaoService,
+    private val gameDataService: GameDataService,
+    refresherScope: CoroutineScope,
+    private val logger: KLogger,
 ) {
 
+    private val sessionsRefresh = 1.seconds
+    private val liveGamesSessions = mutableListOf<LiveGamesWebSocketSession>()
+
+    private val refreshJob = launchAtFixedRate(
+        scope = refresherScope,
+        initialDelay = sessionsRefresh,
+        period = sessionsRefresh,
+        action = { refreshLiveGamesSessions() }
+    )
+
+    fun cancel() {
+        refreshJob.cancel()
+    }
+
     suspend fun listUpcomingEvents(): GetUpcomingEventsResponse {
         return upcomingEventDaoService
             .listUpcomingEventsForLobby()
@@ -24,4 +51,56 @@ class LobbyService(
             }
     }
 
+    fun startLiveGamesSession(
+        sendCb: (LatestGamesUpdateResponse) -> ChannelResult<Unit>,
+    ): String {
+        val session = LiveGamesWebSocketSession(sendCb)
+        liveGamesSessions.add(session)
+        logger.debug { "created $session" }
+        return session.sessionId
+    }
+
+    fun handleLiveGamesSubscription(sessionId: String, gameIds: List<GameId>) {
+        liveGamesSessions
+            .find { it.sessionId == sessionId }
+            ?.updateSubscription(gameIds)
+    }
+
+    fun closeLiveGamesSession(sessionId: String) {
+        liveGamesSessions
+            .find { it.sessionId == sessionId }
+            ?.markAsClosed()
+    }
+
+    private suspend fun refreshLiveGamesSessions() {
+        // remove the sessions that are not active anymore before doing any work
+        liveGamesSessions.removeIf { session ->
+            if (session.isClosed) logger.debug { "removing $session" }
+            session.isClosed
+        }
+
+        if (liveGamesSessions.isNotEmpty()) {
+            // Lobby sessions watch mostly the same games, so fetch everything once: collect
+            // all watched games and, per game, the lowest tracked move index across sessions
+            // (so the single fetch covers every move any session may still need). Each session
+            // then slices out only the moves it should actually receive.
+            val requests = liveGamesSessions.map { session -> session.currentRequest() }
+            val allGameIds = requests.flatMap { request -> request.gameIds }.distinct()
+
+            if (allGameIds.isNotEmpty()) {
+                val batchMoveIndexes = mutableMapOf<String, Int>()
+                requests.forEach { request ->
+                    request.moveIndexes.forEach { (gameId, index) ->
+                        batchMoveIndexes.merge(gameId, index, ::minOf)
+                    }
+                }
+
+                val request = LatestGamesUpdateRequest(allGameIds, batchMoveIndexes)
+                val response = gameDataService.fetchLatestGamesUpdate(request)
+                val batchUpdate = LiveGamesBatchUpdate(response, batchMoveIndexes)
+                liveGamesSessions.forEach { session -> session.update(batchUpdate) }
+            }
+        }
+    }
+
 }

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/services/ws/LiveGamesWebSocketSession.kt

@@ -0,0 +1,103 @@
+package io.elephantchess.servicelayer.services.ws
+
+import io.elephantchess.model.GameEventType
+import io.elephantchess.model.GameId
+import io.elephantchess.servicelayer.dto.lobby.LatestGamesUpdateRequest
+import io.elephantchess.servicelayer.dto.lobby.LatestGamesUpdateResponse
+import kotlinx.coroutines.channels.ChannelResult
+
+/**
+ * Batched update handed to every live games session: the [response] is fetched once
+ * for the union of all watched games (using the lowest tracked move index per game,
+ * see [batchMoveIndexes]) so each session can slice out only the moves it still needs.
+ */
+data class LiveGamesBatchUpdate(
+    val response: LatestGamesUpdateResponse,
+    val batchMoveIndexes: Map<String, Int>,
+)
+
+/**
+ * WebSocket session for a lobby client watching the live games thumbnails.
+ *
+ * The client declares which games it watches via [updateSubscription]; the last
+ * known move index per game is tracked here (not sent by the client), so the
+ * refresher can request only the new moves and the session only pushes entries
+ * that actually changed (new moves or a status change).
+ */
+class LiveGamesWebSocketSession(
+    private val sendCb: (LatestGamesUpdateResponse) -> ChannelResult<Unit>,
+) : WebSocketSession<LiveGamesBatchUpdate>() {
+
+    private var subscribedGameIds: List<GameId> = emptyList()
+    private val moveIndexes = mutableMapOf<String, Int>()
+    private val lastStatuses = mutableMapOf<String, GameEventType>()
+
+    fun updateSubscription(gameIds: List<GameId>) {
+        subscribedGameIds = gameIds
+
+        // forget tracking for games that are not watched anymore
+        val watchedIds = gameIds.map { it.id }.toSet()
+        moveIndexes.keys.retainAll(watchedIds)
+        lastStatuses.keys.retainAll(watchedIds)
+    }
+
+    fun currentRequest(): LatestGamesUpdateRequest =
+        LatestGamesUpdateRequest(
+            gameIds = subscribedGameIds,
+            moveIndexes = moveIndexes.toMap()
+        )
+
+    override fun update(update: LiveGamesBatchUpdate) {
+        val watchedIds = subscribedGameIds.map { it.id }.toSet()
+        val entries = update.response.entries
+            .filter { entry -> entry.gameId.id in watchedIds }
+            .map { entry -> sliceForSession(entry, update.batchMoveIndexes) }
+
+        val changedEntries = entries.filter { entry ->
+            val id = entry.gameId.id
+            val firstTime = id !in lastStatuses
+            val statusChanged = lastStatuses[id] != entry.status
+            firstTime || statusChanged || entry.newMoves.isNotEmpty()
+        }
+
+        if (changedEntries.isNotEmpty()) {
+            val result = sendCb(LatestGamesUpdateResponse(changedEntries))
+            if (result.isClosed) {
+                markAsClosed()
+                return
+            } else if (result.isFailure) {
+                logger.error { "failed to send data to live games session $sessionId" }
+                return
+            }
+        }
+
+        // advance the tracked indexes/statuses so the next request only asks for newer moves
+        entries.forEach { entry ->
+            entry.moveIndex?.let { moveIndexes[entry.gameId.id] = it }
+            lastStatuses[entry.gameId.id] = entry.status
+        }
+    }
+
+    /**
+     * The batched moves start at [LiveGamesBatchUpdate.batchMoveIndexes] (the lowest index
+     * any session needed), so drop the moves this session has already seen. On the first
+     * update for a game (no tracked index yet) no moves are animated: the client loads the fen.
+     */
+    private fun sliceForSession(
+        entry: LatestGamesUpdateResponse.Entry,
+        batchMoveIndexes: Map<String, Int>,
+    ): LatestGamesUpdateResponse.Entry {
+        val tracked = moveIndexes[entry.gameId.id]
+        val batchFrom = batchMoveIndexes[entry.gameId.id]
+        val newMoves = when {
+            tracked == null -> emptyList()
+            batchFrom == null -> entry.newMoves
+            else -> entry.newMoves.drop(maxOf(0, tracked - batchFrom))
+        }
+        return entry.copy(newMoves = newMoves)
+    }
+
+    override fun toString() =
+        "${javaClass.simpleName}{sessionId=$sessionId, games=${subscribedGameIds.size}}"
+
+}

webapp-service-layer/src/main/kotlin/io/elephantchess/servicelayer/utils/ShutdownHandler.kt

@@ -2,6 +2,7 @@ package io.elephantchess.servicelayer.utils
 
 import io.elephantchess.engines.EnginePool
 import io.elephantchess.servicelayer.batch.definitions.BatchesScheduler
+import io.elephantchess.servicelayer.services.LobbyService
 import io.elephantchess.servicelayer.services.PlayerVsBotGameService
 import io.elephantchess.servicelayer.services.PlayerVsPlayerGameService
 import io.elephantchess.servicelayer.services.PuzzleCache
@@ -26,6 +27,7 @@ class ShutdownHandler(
     private val enginePool: EnginePool,
     private val batchesScheduler: BatchesScheduler,
     private val siteMapService: SiteMapService,
+    private val lobbyService: LobbyService,
     private val coroutineScope: CoroutineScope,
 ) {
 
@@ -82,6 +84,14 @@ class ShutdownHandler(
             logger.error(e) { "error cancelling batch scheduler" }
         }
 
+        try {
+            // cancel LobbyService
+            logger.info { "cancelling LobbyService..." }
+            lobbyService.cancel()
+        } catch (e: Exception) {
+            logger.error(e) { "error cancelling LobbyService" }
+        }
+
         try {
             // close engine pool
             logger.info { "closing engine pool..." }